package bp;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;
import javax.persistence.Query;

import job.Producer.Topic;
import job.ProducerFactory;
import model.Behindschedule;
import model.Departpassenger;
import model.DepartpassengerDel;
import model.Errorticket;
import model.ExchangefeeInfo;
import model.FullReturnSchedulePrice;
import model.Insurance;
import model.InsuranceCancel;
import model.InsuranceReturn;
import model.InterfaceInsurance;
import model.Order;
import model.Outlets;
import model.Qhstationdata;
import model.Refundexchangefee;
import model.ScheduleCancel;
import model.Scheduleprice;
import model.Staff;
import model.Station;
import model.StationWebTicket;
import model.Stationcanceldetail;
import model.Stationreturndetail;
import model.Stationselldetail;
import model.Ticketcancel;
import model.Ticketreturn;
import model.Ticketsell;
import model.Tkusedetail;
import model.Unit;
import model.Vehicle;
import model.vo.Outletsinfo;
import model.vo.Selfsaleticket;
import model.vo.Stationbehindschedule;
import model.vo.Stationstaff;
import model.vo.Ticketsources;
import net.sf.json.JSONObject;

import org.codehaus.jackson.JsonNode;

import param.pub.PayRecord;
import param.pub.ScheduleCancelinfo;
import param.secondStage.CountQueryResponse;
import param.secondStage.NetOrganization;
import param.secondStage.StationInsurance;
import param.secondStage.Stationsend;
import param.secondStage.Stationviewgps;
import param.secondStage.Ticketlossdetail;
import play.Logger;
import play.db.jpa.JPA;
import service.DataSynService;
import utils.FNVHash;
import utils.JsonUtil;
import utils.Nodepartdate;
import utils.NovaDateUtil;
import utils.RedisUtil;
import utils.RedisUtil.RedisDBApp;
import utils.SystemParam;
import utils.TopicUtil;
import utils.commons.CommonUtils;
import utils.commons.CommonUtils.TableIndex;
import utils.commons.DateUtils;
import utils.commons.StringUtils;

public class DataSynBP implements DataSynService {
	final private org.apache.log4j.Logger log = Logger.log4j;
	private static SimpleDateFormat sdaf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SS");
	private static SimpleDateFormat datetimeformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	private static SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd");
	/*@Override
	public CountQueryResponse countquery(int orgid, String interfacename) {
		try {
			CountQueryRequest countQueryRequest = new CountQueryRequest();
			CountQueryResponse countQueryResponse = null;
			// 取中心存储的该车站上次成功同步时间
			Datasynctime datasynctime = getDatasynctime(orgid, interfacename);
			Date lastsynctime = DateUtils.StrToDate("2010-01-01");
			if (datasynctime != null) {
				lastsynctime = (Date) datasynctime.getLastsynctime();
			}

			SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			String serversaddress = getServersAddress(orgid);

			countQueryRequest.setInterfacename(interfacename);
			countQueryRequest.setLastsyntime(format.format((lastsynctime)));
			countQueryRequest.setPublicrequest(PublicJsonUtil.newPublicrequest());

			String ORG_SQL = "select code from organization o where o.id=" + orgid;
			Query queryorg = JPA.em().createNativeQuery(ORG_SQL);
			String code = null;
			try {
				code = queryorg.getSingleResult().toString();
			} catch (Exception e) {
				log.error("找不到" + orgid + "机构信息", e);
				return null;
			}
			countQueryRequest.getPublicrequest().setTargetorgcode(code);

			String josnrequest = JsonUtil.parseObject(countQueryRequest);
			String json = "{" + requestful(serversaddress, "/countquery", josnrequest);
			countQueryResponse = (CountQueryResponse) JsonUtil.readJson2Entity(json, CountQueryResponse.class);
			if ("0000".equals(countQueryResponse.getPublicresponse().getResultcode())) {
				return countQueryResponse;
			} else {
				return null;
			}
		} catch (Exception e) {
			log.error("orgid为" + orgid + "中心同步数据量查询失败", e);
			return null;
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public void orgsyn(int orgid) {
		try {
			OrgSyncRequest request = new OrgSyncRequest();
			OrgSyncResponse response = null;
			request.setPublicrequest(PublicJsonUtil.newPublicrequest());

			String CODE_SQL = "select code from organization o where o.id=" + orgid;
			Query querycode = JPA.em().createNativeQuery(CODE_SQL);
			String code = null;
			try {
				code = querycode.getSingleResult().toString();
			} catch (Exception e) {
				log.error("找不到" + orgid + "机构信息", e);
				return;
			}
			request.getPublicrequest().setTargetorgcode(code);

			try {

				// 取中心存储的该车站上次成功同步时间
				Datasynctime datasynctime = getDatasynctime(orgid, "orgsyn");
				Date lastsynctime = DateUtils.StrToDate("2010-01-01");
				if (datasynctime != null) {
					lastsynctime = (Date) datasynctime.getLastsynctime();
				}

				// 根据时间取要同步的机构
				String ORG_SQL = "select * from organization o where o.isactive=1";
				Query queryorg = JPA.em().createNativeQuery(ORG_SQL, Organization.class);
				List<Organization> organizations = queryorg.getResultList();

				String serversaddress = getServersAddress(orgid);
				SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
				if (organizations != null && organizations.size() > 0) {
					List<Org> orgs = new ArrayList<Org>();
					for (Organization organization : organizations) {
						Org org = new Org();
						org.setAddress(organization.getAddress());
						org.setCity(organization.getCity());
						org.setLinkman(organization.getLinkman());
						org.setOrgcode(organization.getCode());
						org.setOrgname(organization.getName());
						org.setOrgtype(organization.getType());
						org.setPhoneno(organization.getPhoneno());
						org.setStationlvl(organization.getStationlvl());
						org.setTrafficroute(organization.getTrafficroute());
						org.setUpdatetime(sdf.format(organization.getUpdatetime()));
						orgs.add(org);
					}
					request.setLastsynctime(lastsynctime);
					request.setOrgs(orgs);

					// 取当前时间为startsynctime
					Date startsynctime = DateUtils.add(new Date(), Calendar.SECOND, -3);

					// 发送请求（返回json字符串）
					String josnrequest = JsonUtil.parseObject(request);
					String json = "{" + requestful(serversaddress, "/orgsyn", josnrequest);
					response = (OrgSyncResponse) JsonUtil.readJson2Entity(json, OrgSyncResponse.class);

					// 判断返回结果，如果返回码为0000，则更新该车站成功同步时间
					Publicresponse publicresponse = response.getPublicresponse();
					if ("0000".equals(publicresponse.getResultcode())) {
						SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
						updatelastsynctime(datasynctime, format.format(startsynctime));
						log.info(orgid + "机构同步成功");
					} else {
						log.error(orgid + "机构同步失败");
					}
				}
			} catch (Exception e) {
				log.error(orgid + "机构同步失败", e);
			}
		} catch (Exception e) {
			log.error(orgid + "机构同步失败", e);
		} finally {
			Scansyncdatajob.putMap(Scansyncdatajob.map, orgid + "_orgsynjob", null);
			// Scansyncdatajob.map.put(orgid + "_orgsynjob", false);
			updatelastruntime("orgsynjob", orgid);
		}

	}*/

	@Override
	public void stationsyn(List<String> messages) {
		if(!checkCanConsume(Topic.Station.name())){
			return;
		}
		log.info("station数据开始消费-----------"+messages.size()+"  条----------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			String hashids;
			StringBuffer hashidbuffer = new StringBuffer();
			List<Station> stations = new ArrayList<Station>();
			Date createtime = new Date();
			JsonNode jnode = null;
			String content = null;
			JSONObject json = null;
			for (String message : messages) {
				Long stationhashid = null;
				try {
					jnode = JsonUtil.getJsonNode(message);
					content = jnode.get("content").toString();
					json = JSONObject.fromObject(content);
					stationhashid = json.getLong("hashid");
					
					Station station = new Station();
					station.id = CommonUtils.GetPKID(TableIndex.station);
					station.city = json.getString("city");
					station.code = json.getString("code");
					station.county = json.getString("county");
					station.createby = 0l;
					station.createtime = createtime;
					station.extendinfo = json.getString("extendinfo");
					station.hashid = json.getLong("hashid");
					station.isactive = json.getBoolean("isactive");
					station.name = json.getString("name");
					station.orgcode = json.getString("orgcode");
					station.province = json.getString("province");
					station.secondname = json.getString("secondname");
					station.stationlvl = json.getString("stationlvl");
					station.updateby = 0l;
					station.updatetime = createtime;
					stations.add(station);
					hashidbuffer.append(stationhashid).append(",");
//					msghashidbuffer.append(FNVHash.hash64(content)).append(",");
				} catch (Exception e) {
					log.error(stationhashid + "站点同步失败1", e);
					flag = false;
				}
			}		
			hashids = hashidbuffer.toString();
			if(hashids.endsWith(","))
				hashids = hashids.substring(0, hashids.length()-1);
			String DEL_STATION_SQL=null;
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				DEL_STATION_SQL = "delete from station  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_STATION_SQL).executeUpdate();
				Map<Long,Station> alreadyIns = new HashMap<Long,Station>();
				for (Station newstation : stations) {
					if(alreadyIns.get(newstation.hashid) != null){
						continue;
					}else{
						alreadyIns.put(newstation.hashid, newstation);
					}
					JPA.em().persist(newstation);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化站点数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
			
				
		}catch(Exception e){
			log.error( "站点同步失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Station), messages);
			}
		}
		log.info("station消费结束----------------------------"+sdaf.format(new Date()));
	}

	/**
	 * 判断该主题是否需要消费
	 * @param name
	 * @return
	 */
	private boolean checkCanConsume(String name) {
		String unconsumetopic = SystemParam.TOPIC_NOT_CONSUME;
		boolean result = true;
		if(unconsumetopic !=null && !"".equals(unconsumetopic)){
			String[] topics = unconsumetopic.split(",");
			for(String topic : topics){
				if(topic.trim().equals(name.trim())){
					result=false;
					break;
				}
			}
		}
		return result;
	}

	@Override
	public CountQueryResponse countquery(int orgid, String interfacename) {
		return null;
	}

	@Override
	public void orgsyn(int orgid) {
		
	}


	public void pricesyn(List<String> messages) {
		if(!checkCanConsume(Topic.SchedulePrice.name())){
			return;
		}
		log.info("班次数据开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			
			String hashids;
			StringBuffer hashidbuffer = new StringBuffer();
			List<Scheduleprice> prices = new ArrayList<Scheduleprice>();
			StringBuffer msghashidbuffer = new StringBuffer();
			List<String> delkeys = new ArrayList<String>();
			Map<String,String> pricedelmap = new HashMap<String, String>();
			Map<String,String> redisresult = new HashMap<String,String>();
			for (String message : messages) {//将所有数据的hashid放入delkeys中，用于下面redis的批量查询
				String pricehashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					pricehashid = contentnodes.get("hashid").asText();
					delkeys.add(pricehashid);
				} catch (Exception e) {
					log.error(pricehashid + "班次计划同步失败2", e);
					flag = false;
				}
			}	
			//批量查询redis
			redisresult = RedisUtil.get(RedisDBApp.pricedel, delkeys);
			for (String message : messages) {
				Long pricehashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					pricehashid = contentnodes.get("hashid").asLong();
					if(redisresult.get(contentnodes.get("hashid").asText())!=null){
						if(Long.valueOf(redisresult.get(contentnodes.get("hashid").asText()))
								>contentnodes.get("updatetime").asLong()){
							continue;
						}else{
							pricedelmap.put(pricehashid.toString(), contentnodes.get("updatetime").asText());
						}
					}
					Scheduleprice price = new Scheduleprice(CommonUtils.GetPKID("scheduleprice"),
							contentnodes.get("hashid").asLong(), contentnodes.get("areatype").asText(), 
							contentnodes.get("busbrand").asText(), contentnodes.get("cansell").asBoolean(),
							0l, new Date(), 
							NovaDateUtil.getDateFromString(contentnodes.get("departdate").asText(), "yyyy-MM-dd"), 
							contentnodes.get("departtime").asText(),contentnodes.get("discountprice").asDouble(), 
									contentnodes.get("endstationcode").asText(),contentnodes.get("endstationname").asText(),		
							contentnodes.get("endtime").asText(),contentnodes.get("fullprice").asDouble(),
							contentnodes.get("halfprice").asDouble(), contentnodes.get("islinework").asBoolean(), 
							contentnodes.get("isovertime").asBoolean(),
							contentnodes.get("orgcode")==null?null:contentnodes.get("orgcode").asText(),
							contentnodes.get("reachstationcode").asText(),contentnodes.get("reachstationname").asText(),
							contentnodes.get("remainkidseats").asInt(),
							contentnodes.get("remainseats").asInt(),
							contentnodes.get("rundistance").asInt(), contentnodes.get("runtime").asDouble(), 
							contentnodes.get("schedulecode").asText(), contentnodes.get("scheduletype").asText(), 
							contentnodes.get("schstatus").asText(),contentnodes.get("seattype").asText() , 
							contentnodes.get("starttime").asText(), contentnodes.get("studentprice").asDouble(),
							contentnodes.get("ticketentrance").asText(),contentnodes.get("topprice").asDouble(), 
							contentnodes.get("totalseats").asInt(), 0l,new Date(contentnodes.get("updatetime").asLong()), 
							contentnodes.get("vehiclelvl").asText(),contentnodes.get("vehicleno").asText(),
							contentnodes.get("vehicletype").asText(),contentnodes.get("waitingroom").asText(),
							contentnodes.get("extendinfo")==null?null:contentnodes.get("extendinfo").asText(),contentnodes.get("routecode")==null?null:contentnodes.get("routecode").asText(),
							contentnodes.get("routename")==null?null:contentnodes.get("routename").asText(),
							contentnodes.get("startstationname")==null?null:contentnodes.get("startstationname").asText(),
							contentnodes.get("venue")==null?null:contentnodes.get("venue").asText(),
							contentnodes.get("ticketprice")==null?null:contentnodes.get("ticketprice").asDouble(),
							contentnodes.get("direction")==null?null:contentnodes.get("direction").asText(),
							contentnodes.get("stopnames")==null?null:contentnodes.get("stopnames").asText(),
							contentnodes.get("rebateprice")==null?null:contentnodes.get("rebateprice").asDouble(),
							contentnodes.get("returnprice")==null?null:contentnodes.get("returnprice").asDouble(),
							contentnodes.get("tripprice")==null?null:contentnodes.get("tripprice").asDouble(),
							contentnodes.get("isrealname")==null?null:contentnodes.get("isrealname").asText(),
							contentnodes.get("temp1")==null?null:contentnodes.get("temp1").asText(),
							contentnodes.get("temp2")==null?null:contentnodes.get("temp2").asText(),
							contentnodes.get("priceversion")==null?null:contentnodes.get("priceversion").asText());
//					System.out.println("hashid : "+price.hashid+" -orgcode : "+price.orgcode+"  -departdate : "+price.departdate.getTime()
//							+" -schedulecode : "+price.schedulecode+" -reachstationcode : "+price.reachstationcode+" -seattype : "+price.seattype
//							+" -newHashid : "+FNVHash.hash64(price.orgcode+price.departdate.getTime()+price.schedulecode+price.reachstationcode+price.seattype));
					price.extendinfo = contentnodes.get("extendinfo")==null?null:contentnodes.get("extendinfo").asText();
					prices.add(price);
					hashidbuffer.append(pricehashid).append(",");
					msghashidbuffer.append(FNVHash.hash64(content)).append(",");
				} catch (Exception e) {
					log.error(pricehashid + "班次计划同步失败1", e);
					flag = false;
				}
			}		
			hashids = hashidbuffer.toString();
			if(hashids.endsWith(","))
				hashids = hashids.substring(0, hashids.length()-1);
			if(StringUtils.isNotEmptyString(hashids)){
				Map<Long,Scheduleprice> alreadyIns = new HashMap<Long,Scheduleprice>();
				try {
					if (!JPA.em().getTransaction().isActive()) {//开启事务
						JPA.em().getTransaction().begin();
					}
					String DEL_SCHEDULEPRICE_SQL = "delete from scheduleprice  " +
							" where hashid in ("+hashids+")";
					JPA.em().createNativeQuery(DEL_SCHEDULEPRICE_SQL).executeUpdate();
					JPA.em().getTransaction().commit();
					if (!JPA.em().getTransaction().isActive()) {//开启事务
						JPA.em().getTransaction().begin();
					}
					for (Scheduleprice scheduleprice : prices) {
						if(alreadyIns.get(scheduleprice.hashid) != null){
							continue;
						}else{
							alreadyIns.put(scheduleprice.hashid, scheduleprice);
						}
						if(scheduleprice.areatype==null || "".equals(scheduleprice.areatype)){
							scheduleprice.areatype = "0";
						}
						if(scheduleprice.departtime!=null && scheduleprice.departtime.length()<6){
							JPA.em().persist(scheduleprice);
						}else{
							log.info("垃圾消息：" + JsonUtil.parseObject(scheduleprice));
						}
					}
					JPA.em().getTransaction().commit();
					JPA.em().clear();
					//批量插入redis
					RedisUtil.set(RedisDBApp.pricedel, pricedelmap);
				} catch (Exception e) {
					log.error("持久化班次计划数据失败2", e);
					if (!JPA.em().getTransaction().isActive()) {//开启事务
						JPA.em().getTransaction().begin();
					}
					JPA.em().getTransaction().rollback();
					flag = false;
				}
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
			
		}catch(Exception e){
			log.error("持久化班次计划数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.SchedulePrice), messages);
			}
		}
		log.info("班次数据消费结束---------------------------"+sdaf.format(new Date()));
	}

	@SuppressWarnings("unchecked")
	@Override
	public void ticketsyn(List<String> messages) {
		if(!checkCanConsume(Topic.Ticket.name())){
			return;
		}
		/*messages =new ArrayList<String>();
		String message1="{\"msgtime\":1527309404177,\"from\":\"center\",\"content\":{\"departdate\":\"2018-05-26\",\"netticketid\":\"a740e0db-3704-4180-bdf4-ddd2d3425c83\",\"returnfee\":0.0,\"returnrate\":0.0,\"returntime\":1527307079000,\"takestatus\":\"1\",\"taketime\":null,\"ticketno\":\"20180526104228482\",\"ticketstatus\":\"1\",\"returnvoucherno\":\"20180526104228482\",\"insuranceno\":null,\"extendinfo\":\"\",\"returnway\":\"2\",\"returnfeemode\":\"1\",\"returnorgcode\":\"11011SYZXZ\",\"returntype\":\"5\",\"cancelfeemode\":null,\"cancelorgcode\":null,\"cancelway\":null,\"canceltime\":null,\"updatetime\":1527309404177,\"insurancestatus\":null,\"returninsuranceorgcode\":null,\"returninsuranceorgname\":null,\"returninsurancetime\":null,\"returninsuranceway\":null,\"cancelinsuranceorgcode\":null,\"cancelinsuranceorgname\":null,\"cancelinsurancetime\":null,\"cancelinsuranceway\":null,\"orderno\":\"78250181039\"}}";
		messages.add(message1);*/
		log.info("电子票开始消费------"+messages.size()+"  条------------"+sdaf.format(new Date()));
		//错误消息，用于存储消费失败的消息，重新发送到kafka
		List<String> errormessages = new ArrayList<String>();
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			StringBuffer msghashidbuffer = new StringBuffer();
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				Map<Long,Ticketreturn> sellids = new HashMap<Long,Ticketreturn>();
				Map<String,Insurance> insurMap = new HashMap<String,Insurance>();
				Map<String,InsuranceReturn> insureturnMap = new HashMap<String,InsuranceReturn>();
				Map<String,InsuranceCancel> insucancelMap = new HashMap<String,InsuranceCancel>();
				StringBuilder sb = new StringBuilder();
				for (String message : messages) {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					sb.append("'").append(contentnodes.get("netticketid").asText()).append("',");
				}
				String hashids = sb.toString();
				hashids = hashids.substring(0, hashids.length()-1);
				String sql = "select t from Ticketsell t where  t.netticketid in ("+hashids+") ";
				Query query = JPA.em().createQuery(sql,Ticketsell.class);
				List<Ticketsell> ticketsells = query.getResultList();
				Map<String,Ticketsell> ticketMap = new HashMap<String,Ticketsell>();
				for(Ticketsell ts : ticketsells){
					ticketMap.put(ts.netticketid, ts);
				}
				List<Insurance> insurances = JPA.em().createQuery("select i from Insurance i where i.netticketid in ("+hashids+")", Insurance.class).getResultList();
				List<InsuranceReturn> insurancereturns = JPA.em().createQuery("select i from InsuranceReturn i where i.netticketid in ("+hashids+")", InsuranceReturn.class).getResultList();
				List<InsuranceCancel> insurancecancels = JPA.em().createQuery("select i from InsuranceCancel i where i.netticketid in ("+hashids+")", InsuranceCancel.class).getResultList();
				for(Insurance i : insurances){
					insurMap.put(i.netticketid, i);
				}
				for(InsuranceReturn ir : insurancereturns){
					insureturnMap.put(ir.netticketid, ir);
				}
				for(InsuranceCancel ic : insurancecancels){
					insucancelMap.put(ic.netticketid, ic);
				}
				for (String message : messages) {
					try {
						if (!JPA.em().getTransaction().isActive()) {//开启事务
							JPA.em().getTransaction().begin();
						}
						JsonNode jnode = JsonUtil.getJsonNode(message);
						String content = jnode.get("content").toString();
						JsonNode contentnodes = JsonUtil.getJsonNode(content);
						String netticketid = contentnodes.get("netticketid").asText();
						/*Station station = new Station(CommonUtils.GetPKID(TableIndex.station), 
								contentnodes.get("city").asText(), contentnodes.get("code").asText(), 
								contentnodes.get("county").asText(), 0l, 
								new Date(), contentnodes.get("hashid").asLong(), contentnodes.get("isactive").asBoolean(),
								null, contentnodes.get("name").asText(), contentnodes.get("orgcode").asText(), 
								orgcache.getOrgid(contentnodes.get("orgcode").asText()),
								contentnodes.get("province").asText(), contentnodes.get("secondname").asText(),
								contentnodes.get("stationlvl").asText(), 0l, new Date());*/
						Ticketsell ticketsell = ticketMap.get(contentnodes.get("netticketid").asText());
						if(ticketsell==null){
							log.error("未找到电子票号为  "+contentnodes.get("netticketid").asText()+"的车票信息。");
							errormessages.add(message);
							flag = false;
							continue;
						}
						Long updatetime = contentnodes.get("updatetime").asLong();
						if(updatetime > ticketsell.updatetime.getTime()){
							Date newdepartdateD = DateUtils.StrToDate(contentnodes.get("departdate").toString().replace("\"", ""));
							String newdepartdate = contentnodes.get("departdate").toString().replace("\"", "");
							
							String olddepartdate = DateUtils.DateToStr(ticketsell.departdate);
							log.info("olddepartdate:"+olddepartdate+"-----newdepartdate"+newdepartdate);
							if(!newdepartdate.equals(olddepartdate)&&newdepartdate!=olddepartdate&&newdepartdate!=null&&!"".equals(newdepartdate)){
								Nodepartdate.setdepartdate(ticketsell.netticketid, ticketsell.departdate, newdepartdateD);
							}
						}
						if(updatetime > ticketsell.updatetime.getTime()){
							String TICKET_UPDATE_SQL = "update ticketsell set "+
									"ticketno='"+contentnodes.get("ticketno").asText()+"',ticketstatus='";
							//废票不处理，当正常票结算==2016-09-21==
							if(!"2".equals(contentnodes.get("ticketstatus").asText())){
								TICKET_UPDATE_SQL += contentnodes.get("ticketstatus").asText()+"',";
							}
							TICKET_UPDATE_SQL += "taketime= :taketime"+"," + "takestatus='"+contentnodes.get("takestatus").asText()+"'," +
								"updatetime= :updatetime where" +
								" id='"+ticketsell.id+"'";
							Query updatequery = JPA.em().createNativeQuery(TICKET_UPDATE_SQL);
							if(!"1".equals(contentnodes.get("takestatus").asText())){
								if(contentnodes.has("taketime")){
									updatequery.setParameter("taketime", new Timestamp(contentnodes.get("taketime").asLong()));
								}else{
									updatequery.setParameter("taketime", null);
								}
							}else{
								updatequery.setParameter("taketime", null);
							}
							updatequery.setParameter("updatetime", new Timestamp(updatetime)).executeUpdate();
						}else if ("0".equals(ticketsell.ticketstatus.trim()) && ("1".equals(contentnodes.get("ticketstatus").asText().trim()) || "2".equals(contentnodes.get("ticketstatus").asText().trim()))){
							String TICKET_UPDATE_SQL = "update ticketsell set " +
									"ticketno='"+contentnodes.get("ticketno").asText()+"',ticketstatus='"+contentnodes.get("ticketstatus").asText()+"'," +
									"taketime= :taketime"+"," + "takestatus='"+contentnodes.get("takestatus").asText()+"'," +
									"updatetime= :updatetime where" +
									" id="+ticketsell.id+"";
							Query updatequery = JPA.em().createNativeQuery(TICKET_UPDATE_SQL);
							if(!"1".equals(contentnodes.get("takestatus").asText())){
								updatequery.setParameter("taketime", new Timestamp(contentnodes.get("taketime").asLong()));
							}else{
								updatequery.setParameter("taketime", null);
							}
							updatequery.setParameter("updatetime", new Timestamp(updatetime)).executeUpdate();
							log.warn("消费过时电子票信息---消息时间："+updatetime+"--电子票更新时间："+ticketsell.updatetime.getTime()+"--消息体："+message);
						}else{
							log.warn("电子票信息已过时，不再消费---消息时间："+updatetime+"--电子票更新时间："+ticketsell.updatetime.getTime()+"--消息体："+message);
						}
						if(StringUtils.isNotEmptyString(contentnodes.get("returnway").asText()) && !"null".equals(contentnodes.get("returnway").asText())
								&& "1".equals(contentnodes.get("ticketstatus").asText().trim())){
							Query returnQuery=JPA.em().createNativeQuery("select * from ticketreturn where netticketid=:netticketid", Ticketreturn.class);
							returnQuery.setParameter("netticketid", contentnodes.get("netticketid").asText());
							List<Ticketreturn> ticketreturns=returnQuery.getResultList();
							
							if(ticketreturns.isEmpty()){
								
								Ticketreturn ticketreturn= new Ticketreturn();
								if(sellids.get(ticketsell.id)!=null){
									ticketreturn = sellids.get(ticketsell.id);
								}else{
									sellids.put(ticketsell.id,ticketreturn);
									ticketreturn.setId(ticketsell.id);
								}
								
								ticketreturn.setNetticketid(ticketsell.netticketid);
								ticketreturn.setDepartorgcode(ticketsell.departorgcode);
								ticketreturn.setDepartdate(ticketsell.departdate);
								ticketreturn.setReturnorgcode(contentnodes.get("returnorgcode").asText());
								ticketreturn.setTicketno(ticketsell.ticketno);
								ticketreturn.setTicketprice(new BigDecimal(ticketsell.price));
								ticketreturn.setCharges(contentnodes.get("returnfee").asDouble());
								ticketreturn.setReturnamount((ticketsell.price.doubleValue())-(contentnodes.get("returnfee").asDouble()));
								if(contentnodes.get("returnvoucherno")!=null)
								ticketreturn.setReturnvoucherno(contentnodes.get("returnvoucherno").asText());
								if(contentnodes.get("returnname")!=null){
									ticketreturn.setReturnname(contentnodes.get("returnname").asText());
								}else{
									ticketreturn.setReturnname("");
								}
								if(contentnodes.get("returntime")!=null)
								ticketreturn.setReturntime(new Date(contentnodes.get("returntime").asLong()));
								ticketreturn.setReturnip("127.0.0.1");
								ticketreturn.setTicketoutlets(ticketsell.ticketoutlets);
								ticketreturn.setReturnway(contentnodes.get("returnway").asText());
								ticketreturn.setReturnfeemode(contentnodes.get("returnfeemode").asText());
								ticketreturn.setTurnoverdetailid(0);
								Date date=new Date();
								ticketreturn.setCreatetime(date);
								ticketreturn.setCreateby(0);
								ticketreturn.setUpdatetime(date);
								ticketreturn.setUpdateby(0);
								
								//2018-01-02
								ticketreturn.setReturntype(contentnodes.get("returntype").asText()!="null"?contentnodes.get("returntype").asText():null);
								
								JPA.em().persist(ticketreturn);
								
							}else{
								String setReturnvoucherno = "";
								String setReturntime = "";
								JsonNode returnvoucherno = contentnodes.get("returnvoucherno");
								if(returnvoucherno != null && !"null".equals(returnvoucherno.asText()) && ((returnvoucherno.asText().trim().length()) > 0)){
									setReturnvoucherno=" returnvoucherno='"+returnvoucherno.asText()+"',";
								}
								if(contentnodes.get("returntime") != null && contentnodes.get("returntime").asText().trim().length()>0
										&& !"null".equals(contentnodes.get("returntime").asText().trim())){
									setReturntime = "returntime= :returntime,";
								}
								String REFUND_SYN_UPDATE = "update ticketreturn set "+setReturnvoucherno+"returnway='"+contentnodes.get("returnway").asText()+"',"
							            +"charges='"+contentnodes.get("returnfee").asText()+"',returnfeemode='"+contentnodes.get("returnfeemode").asText()+"',"+setReturntime
										+"returnorgcode='"+contentnodes.get("returnorgcode").asText()+"',returnamount='"+((ticketsell.price.doubleValue())-(contentnodes.get("returnfee").asDouble()))+"' where " +
										" id='"+ticketreturns.get(0).getId()+"'";
								Query reupdateQuery = JPA.em().createNativeQuery(REFUND_SYN_UPDATE);
								if(contentnodes.get("returntime") != null && contentnodes.get("returntime").asText().trim().length()>0
										&& !"null".equals(contentnodes.get("returntime").asText().trim())){
									reupdateQuery.setParameter("returntime", new Timestamp(Long.parseLong(contentnodes.get("returntime").asText())));
								}
								reupdateQuery.executeUpdate();
							} 
							
							//如果退票已过发车日期未结账，修改为对账不正确
							try {
								Date nowDate = new SimpleDateFormat("yyyy-MM-dd").parse(new SimpleDateFormat("yyyy-MM-dd").format(new Date()));
								if(ticketsell.departdate.getTime()<nowDate.getTime()){
									//判断客运站是否发车日期结账(去掉 isenable=1 and isactive =1判断，加唯一索引)
									String accounttypesql="select balancedatetype from autocheckaccountrule where orgcode =:orgcode  ";
									Query accountquery=JPA.em().createNativeQuery(accounttypesql);
									accountquery.setParameter("orgcode", ticketsell.departorgcode);
									List<Byte> balancedatetype = accountquery.getResultList();
									if(balancedatetype !=null && balancedatetype.size()>0 && balancedatetype.get(0) == 1){
										//判断未结账
										String isjiezhangsql="select `status` from balancesource where orgcode =:departorgcode and accountorgcode = :accountorgcode and datadate= :departdate ";
										Query jiezhangQuery = JPA.em().createNativeQuery(isjiezhangsql);
										jiezhangQuery.setParameter("departorgcode", ticketsell.departorgcode);
										jiezhangQuery.setParameter("accountorgcode", ticketsell.accountorgcode);
										jiezhangQuery.setParameter("departdate", ticketsell.departdate);
										List<String> status = jiezhangQuery.getResultList();
										if(status!=null && status.size()>0 && !"2".equals(status.get(0))){
											EntityTransaction entry=JPA.em().getTransaction();
											try {
												//修改对账为对账不正确
												String updatesql="update cbalanceaccount set status=1 where orgcode =:departorgcode and accountorgcode like :accountorgcode and datadate= :departdate and status=2";
												jiezhangQuery = JPA.em().createNativeQuery(updatesql);
												jiezhangQuery.setParameter("departorgcode", ticketsell.departorgcode);
												jiezhangQuery.setParameter("accountorgcode", "%"+ticketsell.accountorgcode+"%");
												jiezhangQuery.setParameter("departdate", ticketsell.departdate);
												jiezhangQuery.executeUpdate();
											} catch (Exception e) {
												Logger.applogErrorDetail(e, ticketsell.orderno,"车票同步", false, "修改已过发车日期未结账退票为对账不正确", null, null, "修改已过发车日期未结账退票为对账不正确异常");
											}
										}
									}
								}
							} catch (Exception e) {
								Logger.applogErrorDetail(e, ticketsell.orderno,"车票同步", false, "处理已过发车日期未结账退票", null, null, "处理已过发车日期未结账退票异常");
							}
							
						}//废票不处理，当正常票结算==2016-09-21==
						else if(StringUtils.isNotEmptyString(contentnodes.get("ticketstatus").asText()) && !"null".equals(contentnodes.get("ticketstatus").asText())&&"2".equals(contentnodes.get("ticketstatus").asText())){
//							if(contentnodes.get("canceltime")==null||!StringUtils.isNotEmptyString(contentnodes.get("canceltime").asText())){
//								//车站实现为处理废票状态
//								log.error(ticketsell.departorgcode+"车站有废票状态的车票，但未处理废票:"+ticketsell.netticketid+" 。放弃插入废票记录！！！");
//								continue;
//							}
//							//如果是废票状态，且中心库不存在废票记录，要插入TicketCancel表
//							Query cancelQuery=JPA.em().createNativeQuery("select * from ticketcancel where netticketid=:netticketid", Ticketcancel.class);
//							cancelQuery.setParameter("netticketid", contentnodes.get("netticketid").asText());
//							List<Ticketcancel> ticketcancels=cancelQuery.getResultList();
//							if(ticketcancels.isEmpty()){
//								Ticketcancel ticketcancel= new Ticketcancel();
//								ticketcancel.setId(ticketsell.id);
//								ticketcancel.setNetticketid(ticketsell.netticketid);
//								ticketcancel.setDepartorgcode(ticketsell.departorgcode);
//								ticketcancel.setDepartdate(ticketsell.departdate);
//								ticketcancel.setCancelorgcode(contentnodes.get("cancelorgcode").asText());
//								String orgNamequery="select name from organization  where 1=1 and code='"+contentnodes.get("cancelorgcode").asText()+"' ";
//								Query nameQuery=JPA.em().createNativeQuery(orgNamequery);
//								List orgName=nameQuery.getResultList();
//								if(orgName!=null&&orgName.size()>0){
//									ticketcancel.setCancelticketoutlets(orgName.get(0).toString());
//									ticketcancel.setCancelname(orgName.get(0).toString());
//								}else{
//									ticketcancel.setCancelticketoutlets("");
//									ticketcancel.setCancelname("");
//								}
//								ticketcancel.setCancelway(contentnodes.get("cancelway").asText());
//								ticketcancel.setTicketno(ticketsell.ticketno);
//								ticketcancel.setTicketprice(new BigDecimal(ticketsell.price));
//								ticketcancel.setCanceltime(new Date(contentnodes.get("canceltime").asLong()));
//								ticketcancel.setCancelip("127.0.0.1");
//								ticketcancel.setCancelmode(contentnodes.get("cancelfeemode").asText());
//								ticketcancel.setTurnoverdetailid(0);
//								Date date=new Date();
//								ticketcancel.setCreatetime(date);
//								ticketcancel.setCreateby(0);
//								ticketcancel.setUpdatetime(date);
//								ticketcancel.setUpdateby(0);
//								JPA.em().persist(ticketcancel);
//							}
						}
						Insurance insurance = insurMap.get(netticketid);
						if(StringUtils.isNotEmptyString(contentnodes.get("insuranceno").asText()) && !"null".equals(contentnodes.get("insuranceno").asText())){
							insurance.insuranceno=contentnodes.get("insuranceno").asText();
						}
						if(contentnodes.get("insurancestatus") != null && StringUtils.isNotEmptyString(contentnodes.get("insurancestatus").asText()) && !"null".equals(contentnodes.get("insurancestatus").asText().trim()) && !"3".equals(contentnodes.get("insurancestatus").asText().trim())){
							if("1".equals(contentnodes.get("insurancestatus").asText().trim())){
								if(contentnodes.get("returninsurancetime") != null ||!StringUtils.isNotEmptyString(contentnodes.get("returninsurancetime").asText()) || "null".equals(contentnodes.get("returninsurancetime").asText())){
									continue;
								}
								if(insurance == null){
									throw new Exception("未找到电子票号为  "+netticketid+"的保险信息。");
								}
								insurance.updatetime = new Date();
								insurance.insurancestatus = "1";
								//更新保险状态
//								insurance.save();
								InsuranceReturn insuranceReturn = insureturnMap.get(netticketid);
								if(insuranceReturn == null ){
									insuranceReturn = new InsuranceReturn();
									insuranceReturn.id = insurance.id;
									insuranceReturn.insuranceid = insurance.id;
									insuranceReturn.netticketid = netticketid;
									insuranceReturn.departorgcode = insurance.departorgcode;
									insuranceReturn.departdate = insurance.departdate;
									insuranceReturn.returnorgcode = contentnodes.get("returninsuranceorgcode").asText();
									insuranceReturn.premium = insurance.premium;
									insuranceReturn.returnname = contentnodes.get("returninsuranceorgname").asText();
									insuranceReturn.returntime = sdf.parse(contentnodes.get("returninsurancetime").asText().trim());
									insuranceReturn.returnway = contentnodes.get("returninsuranceway").asText();
									insuranceReturn.returnip = "127.0.0.1";
									insuranceReturn.ticketoutlets = insurance.ticketoutlets;
									insuranceReturn.createtime = new Date();
									insuranceReturn.createby = 0;
									insuranceReturn.updatetime = new Date();
									insuranceReturn.updateby = 0;
								
									JPA.em().persist(insuranceReturn);
								}else{
									JPA.em().createNativeQuery("update insurancereturn set returnorgcode = '"+contentnodes.get("returninsuranceorgcode").asText()+"' ,returnname = '"+contentnodes.get("returninsuranceorgname").asText()+"'"
											+ ",returnway = '"+contentnodes.get("returninsuranceway").asText()+"',returntime = :returntime ,updatetime = :updatetime where id = '"+insuranceReturn.id+"'").setParameter("updatetime", new Date())
											.setParameter("returntime", sdf.parse(contentnodes.get("returninsurancetime").asText().trim())).executeUpdate();
								}
							}else if("2".equals(contentnodes.get("insurancestatus").asText().trim())){
								if(contentnodes.get("cancelinsurancetime") != null ||!StringUtils.isNotEmptyString(contentnodes.get("cancelinsurancetime").asText()) || "null".equals(contentnodes.get("cancelinsurancetime").asText())){
									continue;
								}
								if(insurance == null){
									throw new Exception("未找到电子票号为  "+netticketid+"的保险信息。");
								}
								insurance.updatetime = new Date();
								insurance.insurancestatus = "2";
								//更新保险状态
//								insurance.save();
								InsuranceCancel insuranceCancel = insucancelMap.get(netticketid);
								if(insuranceCancel == null ){
									insuranceCancel = new InsuranceCancel();
									insuranceCancel.id = insurance.id;
									insuranceCancel.insuranceid = insurance.id;
									insuranceCancel.netticketid = insurance.netticketid;
									insuranceCancel.departorgcode = insurance.departorgcode;
									insuranceCancel.departdate = insurance.departdate;
									insuranceCancel.cancelorgcode = contentnodes.get("cancelinsuranceorgcode").asText();
									insuranceCancel.premium = insurance.premium;
									insuranceCancel.cancelname = contentnodes.get("cancelinsuranceorgname").asText();
									insuranceCancel.canceltime = sdf.parse(contentnodes.get("cancelinsurancetime").asText().trim());
									insuranceCancel.cancelway = contentnodes.get("cancelinsuranceway").asText();
									insuranceCancel.cancelip = "127.0.0.1";
									insuranceCancel.ticketoutlets = insurance.ticketoutlets;
									insuranceCancel.createtime = new Date();
									insuranceCancel.createby = 0;
									insuranceCancel.updatetime = new Date();
									insuranceCancel.updateby = 0;
								
									JPA.em().persist(insuranceCancel);
								}else{
									JPA.em().createNativeQuery("update insurancecancel set cancelorgcode = '"+contentnodes.get("cancelinsuranceorgcode").asText()+"' ,cancelname = '"+contentnodes.get("cancelinsuranceorgname").asText()+"'"
											+ ",cancelway = '"+contentnodes.get("cancelinsuranceway").asText()+"',canceltime = :canceltime ,updatetime = :updatetime where id = '"+insuranceCancel.id+"'").setParameter("updatetime", new Date())
											.setParameter("canceltime", sdf.parse(contentnodes.get("cancelinsurancetime").asText().trim())).executeUpdate();
								}
							}
							
						}
						if(insurance!=null){
							JPA.em().persist(insurance);
						}
						msghashidbuffer.append(FNVHash.hash64(content)).append(",");
						
						//单个提交，不用批量提交
						if (!JPA.em().getTransaction().isActive()) {//开启事务
							JPA.em().getTransaction().begin();
						}
						JPA.em().getTransaction().commit();
					} catch (Exception e) {
						log.error("车票信息同步失败1"+e.getMessage(),e);
						flag = false;
						errormessages.add(message);
						if (!JPA.em().getTransaction().isActive()) {//开启事务
							JPA.em().getTransaction().begin();
						}
						JPA.em().getTransaction().rollback();
					}
				}		
				
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化车票数据失败2", e);
				flag = false;
				errormessages = messages;
			}
			String msghashids = msghashidbuffer.toString();
			if(msghashids.trim().length()>0 && flag){
				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
			}
		
		}catch(Exception e){
			log.error("持久化车票数据失败2", e);
			flag = false;
			errormessages = messages;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Ticket), errormessages);
			}
		}
		log.info("电子票消费结束---------------------------------"+sdaf.format(new Date()));
	}

	public void pricedelsyn(List<String> messages) {
		if(!checkCanConsume(Topic.Pricedel.name())){
			return;
		}
		log.info("班次删除开始消费-----------"+messages.size()+"  条----------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			String hashids;
			StringBuffer hashidbuffer = new StringBuffer();
			Map<String, String> pricedelmap = new HashMap<String, String>();
			StringBuffer msghashidbuffer = new StringBuffer();
			List<String> delkeys = new ArrayList<String>();
			Map<String,String> redisresult = new HashMap<String,String>();
			for(String message : messages){//将所有数据的hashid放入delkeys中，用于下面redis的批量查询
				String pricedelhashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					pricedelhashid = contentnodes.get("hashid").asText();
					delkeys.add(pricedelhashid);
				} catch (Exception e) {
					log.error(pricedelhashid + "解析班次删除消息失败1", e);
					flag = false;
				}
			}
			//批量查询redis
			redisresult = RedisUtil.get(RedisDBApp.pricedel, delkeys);
			for (String message : messages) {
				Long pricedelhashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					pricedelhashid = contentnodes.get("hashid").asLong();
					Long deletetime = contentnodes.get("deletetime").asLong();
					if(redisresult.get(contentnodes.get("hashid").asText())==null||
							(deletetime>Long.parseLong(redisresult.get(contentnodes.get("hashid").asText())))){
						pricedelmap.put(pricedelhashid.toString(), contentnodes.get("deletetime").asText());
					}
					hashidbuffer.append(pricedelhashid).append(",");
					msghashidbuffer.append(FNVHash.hash64(content)).append(",");
				} catch (Exception e) {
					log.error(pricedelhashid + "解析班次删除消息失败", e);
					flag = false;
				}
			}	
			hashids = hashidbuffer.toString();
			if(hashids.endsWith(","))
				hashids = hashids.substring(0, hashids.length()-1);
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				//删除数据前，先将数据查出并插入schudulepricedle表
//				String PRE_DEL_QUERY = "select * from scheduleprice  where hashid in ("+hashids+")";
//				Query preQuery = JPA.em().createNativeQuery(PRE_DEL_QUERY, Scheduleprice.class);
//				List<Scheduleprice> scheduleprices = preQuery.getResultList();
//				if(scheduleprices!=null){
//					String DEL_PRICEDEL_SQL = "delete from schedulepricedel  " +
//							" where hashid in ("+hashids+")";
//					JPA.em().createNativeQuery(DEL_PRICEDEL_SQL).executeUpdate();
//					Map<Long,Scheduleprice> alreadyIns = new HashMap<Long,Scheduleprice>();
//					for(Scheduleprice price:scheduleprices){
//						if(alreadyIns.get(price.hashid) != null){
//							continue;
//						}else{
//							alreadyIns.put(price.hashid, price);
//						}
//						Schedulepricedel pricedel=new Schedulepricedel();
//						pricedel.setOrgcode(price.orgcode);
//						pricedel.setSchedulecode(price.schedulecode);
//						pricedel.setDepartdate(price.departdate);
//						pricedel.setSeattype(price.seattype);
//						pricedel.setReachstationcode(price.reachstationcode);
//						pricedel.setReachstationname(price.reachstationname);
//						pricedel.setCreatetime(price.createtime);
//						pricedel.setCreateby(price.createby);
//						pricedel.setUpdatetime(price.updatetime);
//						pricedel.setUpdateby(price.updateby);
//						pricedel.setHashid(price.hashid);
//						JPA.em().persist(pricedel);
//					}
//				}
				String DEL_STATION_SQL = "delete from scheduleprice  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_STATION_SQL).executeUpdate();
//				for (Map.Entry<Long, Long> entry : pricedelmap.entrySet()) {//保存删除信息到数据库
//					RedisUtil.set(RedisDBApp.pricedel, 
//							entry.getKey().toString(), entry.getValue().toString(),60*24*7);
//				}
				JPA.em().getTransaction().commit();
				//批量插入redis
				RedisUtil.set(RedisDBApp.pricedel, pricedelmap);
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("删除班次数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化删除班次数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Pricedel), messages);
			}
		}
		log.info("班次删除消费结束-----------------------"+sdaf.format(new Date()));
	}

	@Override
	public void vehiclesyn(List<String> messages) {
		if(!checkCanConsume(Topic.Vehicle.name())){
			return;
		}
		log.info("车牌数据开始消费--------"+messages.size()+"  条--------------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			String hashids;
			StringBuffer hashidbuffer = new StringBuffer();
			List<Vehicle> vehicles = new ArrayList<Vehicle>();
			StringBuffer msghashidbuffer = new StringBuffer();
			Map<Long,Unit> unitmap = new HashMap<Long,Unit>();
			StringBuilder unithashidsb = new StringBuilder();
			for (String message : messages) {
				Long vehiclehashid = null;
				JsonNode jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				JsonNode contentnodes = JsonUtil.getJsonNode(content);
				try {
					vehiclehashid = contentnodes.get("hashid").asLong();
					hashidbuffer.append(vehiclehashid).append(",");
					Date retiredate = null;
					try{
						retiredate = dateformat.parse(contentnodes.get("retiredate").asText());
					}catch(Exception e){
					}
					Date scrapdate = null;
					try{
						scrapdate = dateformat.parse(contentnodes.get("scrapdate").asText());
					}catch(Exception e){
					}
					Vehicle vehicle = new Vehicle(CommonUtils.GetPKID("vehicle"),
							contentnodes.get("hashid").asLong(),contentnodes.get("busbrand").asText(),
							new Date(),contentnodes.get("orgcode").asText(),
							contentnodes.get("rationseatnum").asInt(), contentnodes.get("seatnum").asInt(), 
							new Date(contentnodes.get("updatetime").asLong()), contentnodes.get("vehiclelvl").asText(), 
							contentnodes.get("vehicleno").asText(), "null".equals(contentnodes.get("vehicletype").asText())?null:contentnodes.get("vehicletype").asText(),
							NovaDateUtil.getDateFromString(contentnodes.get("factorydate").asText(), "yyyy-MM-dd"),
							contentnodes.get("unitname").asText(),contentnodes.get("unitcode").asText(),contentnodes.get("status").asText(),
							contentnodes.get("certificatesno").asText(),retiredate,scrapdate,contentnodes.get("bustype").asText(),contentnodes.get("starlvl").asText());
					//将之前不区分大小写的hashid加入删除中在消费没有索引冲突时可以去掉该部分代码---start----
					hashidbuffer.append(FNVHash.hash64(vehicle.orgcode.trim() +"#"+ vehicle.vehicleno.trim())).append(",");
					//将之前不区分大小写的hashid加入删除中---end------
					vehicles.add(vehicle);
					msghashidbuffer.append(FNVHash.hash64(content)).append(",");
				} catch (Exception e) {
					log.error(vehiclehashid + "车牌数据同步失败1", e);
					flag = false;
				}
				
				try {
					if(contentnodes.get("unitcode") != null && !"null".equals(contentnodes.get("unitcode").asText()) && contentnodes.get("unitcode").asText().trim().length()>0){
						Date updatetime = new Date(contentnodes.get("updatetime").asLong());
						Long unithashid = contentnodes.get("unithashid").asLong();
						unithashidsb.append("'").append(unithashid).append("',");
						Unit unit = unitmap.get(unithashid);
						if(unit == null || unit.updatetime.before(updatetime)){
							unit = new Unit();
							unit.orgcode = contentnodes.get("orgcode").asText();
							unit.code = contentnodes.get("unitcode").asText();
							unit.name = contentnodes.get("unitname").asText();
							unit.fullname = contentnodes.get("fullname").asText();
							unit.contactperson = contentnodes.get("contactperson").asText();
							unit.contactphone = contentnodes.get("contactphone").asText();
							unit.city = contentnodes.get("city").asText();
							unit.address = contentnodes.get("address").asText();
							unit.nature = contentnodes.get("nature").asText();
							unit.updatetime = updatetime;
							unit.hashid = unithashid;
							unit.updateby = "0";
							unit.createtime = new Date();
							unit.createby = "0";
							unitmap.put(unithashid, unit);
						}
					}
					
				} catch (Exception e) {
					log.error(vehiclehashid + "车牌数据同步失败3", e);
					flag = false;
				}
			}		
			hashids = hashidbuffer.toString();
			if(hashids.endsWith(","))
				hashids = hashids.substring(0, hashids.length()-1);
			Map<Long,Vehicle> alreadyIns = new HashMap<Long,Vehicle>();
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				String DEL_STATION_SQL = "delete from vehicle  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_STATION_SQL).executeUpdate();
				for (Vehicle vehicle : vehicles) {
					if(alreadyIns.get(vehicle.hashid) != null){
						continue;
					}else{
						alreadyIns.put(vehicle.hashid, vehicle);
					}
					JPA.em().persist(vehicle);
				}
				if(unithashidsb.toString().endsWith(",")){
					String unithashids = unithashidsb.toString().substring(0,unithashidsb.toString().length()-1);
					String DEL_UNIT_SQL = "delete from unit  " +
							" where hashid in ("+unithashids+")";
					JPA.em().createNativeQuery(DEL_UNIT_SQL).executeUpdate();
					for(Unit unit : unitmap.values()){
						JPA.em().persist(unit);
					}
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化车牌数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化车牌数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Vehicle), messages);
			}
		}
		log.info("车牌数据消费结束--------------------------"+sdaf.format(new Date()));
	}

	@Override
	public void departpassengersyn(List<String> messages) {
		if(!checkCanConsume(Topic.Departpassenger.name())){
			return;
		}
		log.info("发班运量开始消费---------"+messages.size()+"  条---------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			String hashids;
			StringBuffer hashidbuffer = new StringBuffer();
			List<Departpassenger> departpassengers = new ArrayList<Departpassenger>();
			//OrganizationCache orgcache = new OrganizationCache();
			StringBuffer msghashidbuffer = new StringBuffer();
			List<String> delkeys = new ArrayList<String>();
			Map<String,String> redisresult = new HashMap<String,String>();
			for (String message : messages) {//将所有数据的hashid放入delkeys中，用于下面redis的批量查询
				String pricehashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					pricehashid = contentnodes.get("hashid").asText();
					delkeys.add(pricehashid);
				} catch (Exception e) {
					log.error(pricehashid + "发班运量同步失败1", e);
					flag = false;
				}
			}
			//批量查询redis
			redisresult = RedisUtil.get(RedisDBApp.pricedel, delkeys);
			for (String message : messages) {
				Long departpassengerhashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					departpassengerhashid = contentnodes.get("hashid").asLong();
					
					if(redisresult.get(contentnodes.get("hashid").asText())!=null){
						if(Long.valueOf(redisresult.get(contentnodes.get("hashid").asText()))
								>contentnodes.get("updatetime").asLong()){
							continue;
						}
					}
					//hashidbuffer.append("'").append(stationhashid).append("'").append(",");
					hashidbuffer.append(departpassengerhashid).append(",");
					Departpassenger departpassener = new Departpassenger(CommonUtils.GetPKID(TableIndex.departpassener), 
							departpassengerhashid,
							new Date(), NovaDateUtil.getDateFromString(contentnodes.get("departdate").asText(), "yyyy-MM-dd"),
							contentnodes.get("isoriginator").asInt(),
							contentnodes.get("moneysellbyotherstation").asDouble(),contentnodes.get("moneysellbyotherway").asDouble(),
							contentnodes.get("moneysellbystation").asDouble(), contentnodes.get("moneysellbywebsite").asDouble(),
							contentnodes.get("numsellbyotherstation").asInt(), contentnodes.get("numsellbyotherway").asInt(), 
							contentnodes.get("numsellbystation").asInt(),contentnodes.get("numsellbywebsite").asInt(),
							contentnodes.get("schedulecode").asText(), new Date(),contentnodes.get("vehicleno").asText(),
							contentnodes.get("rundistance").asInt(), contentnodes.get("orgcode").asText(), contentnodes.get("reachstationcode").asText(), 
							contentnodes.get("reachstationname").asText(),contentnodes.get("stationcharge").asDouble(),contentnodes.get("insurancecount").asInt(),contentnodes.get("insuranceamount").asDouble());
					departpassengers.add(departpassener);
					msghashidbuffer.append(FNVHash.hash64(content)).append(",");
				} catch (Exception e) {
					log.error(departpassengerhashid + "发班运量同步失败1", e);
					flag = false;
				}
			}		
			hashids = hashidbuffer.toString();
			if(hashids.endsWith(","))
				hashids = hashids.substring(0, hashids.length()-1);
			try {
				Map<Long,Departpassenger> alreadyIns = new HashMap<Long,Departpassenger>();
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				String DEL_STATION_SQL = "delete from departpassenger  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_STATION_SQL).executeUpdate();
				for (Departpassenger departpass : departpassengers) {
					if(alreadyIns.get(departpass.hashid) != null){
						continue;
					}else{
						alreadyIns.put(departpass.hashid, departpass);
					}
					JPA.em().persist(departpass);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化发班运量数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化发班运量数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Departpassenger), messages);
			}
		}
		log.info("发班运量消费结束-------------------------------"+sdaf.format(new Date()));
	}

	@SuppressWarnings("unchecked")
	@Override
	public void departpassengerdelsyn(List<String> messages) {
		if(!checkCanConsume(Topic.Departpassengerdel.name())){
			return;
		}
		log.info("发班运量删除开始消费----------"+messages.size()+"  条---------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			String hashids;
			StringBuffer hashidbuffer = new StringBuffer();
			Map<Long, Long> departpassenerdelmap = new HashMap<Long, Long>();
			StringBuffer msghashidbuffer = new StringBuffer();
			for (String message : messages) {
				Long passenerdeldelhashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					passenerdeldelhashid = contentnodes.get("hashid").asLong();
					Long deletetime = contentnodes.get("deletetime").asLong();
					
					hashidbuffer.append(passenerdeldelhashid).append(",");
					msghashidbuffer.append(FNVHash.hash64(content)).append(",");
					if(RedisUtil.get(RedisDBApp.departpassengerdel, contentnodes.get("hashid").asText())==null
							|| (deletetime>Long.parseLong(RedisUtil.get(RedisDBApp.departpassengerdel, contentnodes.get("hashid").asText())))){
						departpassenerdelmap.put(passenerdeldelhashid, contentnodes.get("deletetime").asLong());
					}
				} catch (Exception e) {
					log.error(passenerdeldelhashid + "解析发班运量删除消息失败", e);
					flag = false;
				}
			}	
			hashids = hashidbuffer.toString();
			if(hashids.endsWith(","))
				hashids = hashids.substring(0, hashids.length()-1);
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				//删除之前插入departpassengerdel表
				String PRE_DEL_QUERY = "select *  from departpassenger g where hashid in ("+hashids+")";
				Query preQuery = JPA.em().createNativeQuery(PRE_DEL_QUERY, Departpassenger.class);
				List<Departpassenger> departpassengers = preQuery.getResultList();
				Map<Long,Departpassenger> alreadyIns = new HashMap<Long,Departpassenger>();
				for(Departpassenger departpassenger:departpassengers){
					if(alreadyIns.get(departpassenger.hashid) != null){
						continue;
					}else{
						alreadyIns.put(departpassenger.hashid, departpassenger);
					}
					DepartpassengerDel del = new DepartpassengerDel();
					del.setHashid(departpassenger.hashid);
					del.setOrgcode(departpassenger.orgcode);
					del.setDepartdate(departpassenger.departdate);
					del.setSchedulecode(departpassenger.schedulecode);
					del.setVehicleno(departpassenger.vehicleno);
					del.setReachstationcode(departpassenger.reachstationcode);
					del.setReachstationname(departpassenger.reachstationname);
					del.setCreatetime(departpassenger.createtime);
					del.setUpdatetime(departpassenger.updatetime);
					JPA.em().persist(del);
				}
				//删除过期数据
				String DEL_PASSENER_SQL = "delete from departpassenger  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_PASSENER_SQL).executeUpdate();
				for (Map.Entry<Long, Long> entry : departpassenerdelmap.entrySet()) {//保存删除信息到数据库
					RedisUtil.set(RedisDBApp.departpassengerdel, 
							entry.getKey().toString(), entry.getValue().toString(),60*24*7);
				}
				JPA.em().getTransaction().commit();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化站点数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化站点数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Departpassengerdel), messages);
			}
		}
		log.info("发班运量删除消费结束------------------------"+sdaf.format(new Date()));
	}

	@Override
	public void errorticketsyn(List<String> messages) {
		if(!checkCanConsume(Topic.ErrorTicket.name())){
			return;
		}
		log.info("errorticket开始消费----------"+messages.size()+"  条------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			List<Errorticket> errortickets = new ArrayList<Errorticket>();
			StringBuffer msghashidbuffer = new StringBuffer();
			for (String message : messages) {
				Long errortickethashid = null;
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					JsonNode contentnodes = JsonUtil.getJsonNode(content);
					errortickethashid = contentnodes.get("hashid").asLong();
					//hashidbuffer.append("'").append(stationhashid).append("'").append(",");
					Errorticket errorticket = new Errorticket(CommonUtils.GetPKID("errorticket"),
							contentnodes.get("certificateno").asText(),contentnodes.get("certificatetype").asText(),
							 Long.parseLong("0"), new Date(),contentnodes.get("customername").asText(),
							 NovaDateUtil.getDateFromString(contentnodes.get("departdate").asText(), "yyyy-MM-dd"),
							 contentnodes.get("departorgcode").asText(), contentnodes.get("orderno").asText(), 
							 contentnodes.get("phone").asText(),contentnodes.get("remark").asText(),
							 contentnodes.get("ticketnum").asInt(), Long.parseLong("0"), new Date(), errortickethashid);
					errortickets.add(errorticket);
					msghashidbuffer.append(errortickethashid).append(",");
				} catch (Exception e) {
					log.error(errortickethashid + "取票异常数据同步异常", e);
					flag = false;
				}
			}		
			try {
				String hashids = msghashidbuffer.toString();
				if(hashids.endsWith(","))
					hashids = hashids.substring(0, hashids.length()-1);
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				String DEL_ERRORTICKET_SQL = "delete from errorticket  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_ERRORTICKET_SQL).executeUpdate();
				Map<Long,Errorticket> alreadyIns = new HashMap<Long,Errorticket>();
				for (Errorticket errorticket : errortickets) {
					if(alreadyIns.get(errorticket.hashid) != null){
						continue;
					}else{
						alreadyIns.put(errorticket.hashid, errorticket);
					}
					JPA.em().persist(errorticket);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化取票异常数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化取票异常数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.ErrorTicket), messages);
			}
		}
		log.info("errorticket消费结束-------------------------"+sdaf.format(new Date()));
	}

	
	@Override
	public void selfsaleticket(List<String> messages) {
		if(!checkCanConsume(Topic.SelfSaleTicket.name())){
			return;
		}
		log.info("自助售票车票同步开始消费----------"+messages.size()+"   条----------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			if (!JPA.em().getTransaction().isActive()) {//开启事务
				JPA.em().getTransaction().begin();
			}
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
			Map<String,Order> existsorders = new HashMap<String,Order>();
			Map<String,Ticketsell> existsticketsells = new HashMap<String,Ticketsell>();
			Map<String,Ticketreturn> existsticketreturns = new HashMap<String,Ticketreturn>();
			Map<String,Ticketcancel> existsticketcancels = new HashMap<String,Ticketcancel>();
			StringBuilder ordernosb = new StringBuilder();
			StringBuilder netticketidsb = new StringBuilder();
			for(String message : messages){
				JsonNode jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				Selfsaleticket sst = (Selfsaleticket) JsonUtil.readJson2Entity(content, Selfsaleticket.class);
				ordernosb.append("'").append(sst.getOrderno()).append("',");
				netticketidsb.append("'").append(sst.getNetticketid()).append("',");
			}

			String ordernos = ordernosb.toString().substring(0,ordernosb.toString().length()-1);
			String netticketids = netticketidsb.toString().substring(0, netticketidsb.toString().length()-1);
			List<Order> orderresult = JPA.em().createQuery("select o from Order o where o.orderno in ("+ordernos+")", Order.class)
										.getResultList();
			List<Ticketsell> ticketsellresult = JPA.em().createQuery("select s from Ticketsell s where s.netticketid in ("+netticketids+")", Ticketsell.class)
										.getResultList();
			List<Ticketreturn> ticketreturnresult = JPA.em().createQuery("select r from Ticketreturn r where r.netticketid in ("+netticketids+")", Ticketreturn.class)
					.getResultList();
			List<Ticketcancel> ticketcancelresult = JPA.em().createQuery("select c from Ticketcancel c where c.netticketid in ("+netticketids+")", Ticketcancel.class)
					.getResultList();
			for(Order o : orderresult){
				existsorders.put(o.orderno, o);
			}
			for(Ticketsell t : ticketsellresult){
				existsticketsells.put(t.netticketid, t);
			}
			for(Ticketreturn t : ticketreturnresult){
				existsticketreturns.put(t.getNetticketid(), t);
			}
			for(Ticketcancel t : ticketcancelresult){
				existsticketcancels.put(t.getNetticketid(), t);
			}
			Map<String,Order> alreadyinsertorders = new HashMap<String,Order>();
			Map<String,Ticketsell> alreadyinsertticketsells = new HashMap<String,Ticketsell>();
			Map<String,Ticketreturn> alreadyinsertticketreturns = new HashMap<String,Ticketreturn>();
			
			try{
				for(String message : messages){
					try{
						JsonNode jnode = JsonUtil.getJsonNode(message);
						String content = jnode.get("content").toString();
						if (!JPA.em().getTransaction().isActive()) {//开启事务
							JPA.em().getTransaction().begin();
						}
						Selfsaleticket selfsaleticket = (Selfsaleticket) JsonUtil.readJson2Entity(content, Selfsaleticket.class);
						Order order = existsorders.get(selfsaleticket.getOrderno());
						Date departdate = new SimpleDateFormat("yyyy-MM-dd").parse(selfsaleticket.getDepartdate());
						if(order == null && alreadyinsertorders.get(selfsaleticket.getOrderno()) == null){
							order = new Order();
							order.id = CommonUtils.GetPKID("orders");
							order.orderno = selfsaleticket.getOrderno();
							order.amount = 0.0;
							order.createby = 0;
							order.createtime = sdf.parse(selfsaleticket.getLocktime());
							order.departdate = departdate;
							order.departorgcode = selfsaleticket.getDepartorgcode();
							order.departorgname = selfsaleticket.getDepartorgname();
							order.departtime = selfsaleticket.getDeparttime();
							order.lockdelay = 0;
							order.locktime = sdf.parse(selfsaleticket.getLocktime());
							order.reachstationcode = selfsaleticket.getReachstationcode();
							order.reachstationname = selfsaleticket.getReachstationname();
							order.schedulecode = selfsaleticket.getSchedulecode();
							order.seatnum = selfsaleticket.getSeatnum();
							order.seattype = selfsaleticket.getSeattype();
							order.sellorgcode = selfsaleticket.getSellorgcode();
							order.sellorgname = selfsaleticket.getSellorgname();
							order.sellway = "1";// 网上售票
							order.updateby = 0;
							order.status = selfsaleticket.getStatus();
							order.updatetime = new Date();
							order.save();
							alreadyinsertorders.put(order.orderno, order);
						}
						Ticketsell ts = existsticketsells.get(selfsaleticket.getNetticketid());
						if(ts == null && alreadyinsertticketsells.get(selfsaleticket.getNetticketid()) == null){
							ts = new Ticketsell();

							ts.id = CommonUtils.GetPKID("ticketsell");
							ts.buspark = selfsaleticket.getBuspark();
							ts.certificateno = selfsaleticket.getCertificateno();
							ts.certificatetype = selfsaleticket.getCertificatetype();
							ts.createby = 0l;
							ts.createtime = new Date();
							ts.customername = selfsaleticket.getCustomername();
							ts.departdate = departdate;
							ts.departorgcode = selfsaleticket.getDepartorgcode();
							ts.departtime = selfsaleticket.getDeparttime();
							ts.discountrate = selfsaleticket.getDiscountrate();
							ts.distance = selfsaleticket.getDistance();
							ts.eticketno = selfsaleticket.getEticketno();
							ts.fullprice = selfsaleticket.getFullprice();
							ts.islineschedule = selfsaleticket.getIslineschedule() == 1?true:false;
							ts.takestatus = selfsaleticket.getTakestatus();
							ts.sellway = "1";
							ts.istakekid = selfsaleticket.getIstakekid() == 1 ? true : false;

							ts.orderno = selfsaleticket.getOrderno();
							ts.paymethod = selfsaleticket.getPaymethod();
							ts.phone = selfsaleticket.getPhone();
							ts.price = selfsaleticket.getPrice();
							ts.reachstationcode = selfsaleticket.getReachstationcode();
							ts.schedulecode =selfsaleticket.getSchedulecode();
							ts.netticketid = selfsaleticket.getNetticketid();
							ts.seatno = selfsaleticket.getSeatno();
							ts.seattype = selfsaleticket.getSeattype();
							ts.sellby = 0l;
							ts.sellip = "127.0.0.1";
							ts.sellorgcode = selfsaleticket.getSellorgcode();
							ts.selltime = sdf.parse(selfsaleticket.getSelltime());
							ts.servicefee = selfsaleticket.getServicefee();
							ts.ticketentrance = selfsaleticket.getTicketentrance();
							ts.ticketno = selfsaleticket.getTicketno();
							ts.ticketoutlets = selfsaleticket.getSellorgname();
							//废票不处理，当正常票结算==2017-03-21==
							if(!"2".equals( selfsaleticket.getTicketstatus())){
								ts.ticketstatus = selfsaleticket.getTicketstatus();
							}else{
								ts.ticketstatus = "0";
							}
							ts.tickettype = selfsaleticket.getTickettype();
							ts.toplimitprice = selfsaleticket.getToplimitprice();
							ts.updateby = 0l;
							ts.updatetime = new Timestamp(selfsaleticket.getUpdatetime());
							ts.verificationcode = selfsaleticket.getVerificationcode();
							ts.waitingroom = selfsaleticket.getWaitingroom();
							ts.canreturnmoneyback = selfsaleticket.getCanreturnmoneyback();
							ts.accountorgcode = selfsaleticket.getAccountorgcode();
							ts.equipmentproviders = selfsaleticket.getEquipmentproviders();
							ts.seller = selfsaleticket.getSeller();
							ts.paymerchantname = selfsaleticket.getPaymerchantname();
							ts.paymerchantchannelname = selfsaleticket.getPaymerchantchannelname();
							ts.paymerchantsign = selfsaleticket.getPaymerchantsign();
							ts.customerid = 0L;
							JPA.em().persist(ts);
							alreadyinsertticketsells.put(ts.netticketid, ts);
						}else if(alreadyinsertticketsells.get(selfsaleticket.getNetticketid()) != null){
							
						}else{
							
							Long updatetime = selfsaleticket.getUpdatetime();
							if(updatetime > ts.updatetime.getTime()){
								String UPDATE_TAKETIME_SQL = "";
								if(selfsaleticket.getTaketime() != null ){
									UPDATE_TAKETIME_SQL = "taketime= :taketime,";
								}
								//发车日期不同的话插入混检记录---------------------
								Date newdepartdateD = DateUtils.StrToDate(selfsaleticket.getDepartdate());
								String newdepartdate = selfsaleticket.getDepartdate();
								
								String olddepartdate = DateUtils.DateToStr(ts.departdate);
								log.info("olddepartdate:"+olddepartdate+"-----newdepartdate"+newdepartdate);
								if(!newdepartdate.equals(olddepartdate)&&newdepartdate!=olddepartdate&&newdepartdate!=null&&!"".equals(newdepartdate)){
									Nodepartdate.setdepartdate(ts.netticketid, ts.departdate, newdepartdateD);
								}
								//----------------------------------------
								String TICKET_UPDATE_SQL = "update ticketsell set " +
										"ticketno='"+selfsaleticket.getTicketno()+"'," + UPDATE_TAKETIME_SQL;
								//废票不处理，当正常票结算==2017-03-21==
								if(!"2".equals( selfsaleticket.getTicketstatus())){
									TICKET_UPDATE_SQL += "ticketstatus='"+selfsaleticket.getTicketstatus()+"',";
								}
								TICKET_UPDATE_SQL += "takestatus='"+selfsaleticket.getTakestatus()+"'," +
										"updatetime= :updatetime where" +
										" netticketid='"+selfsaleticket.getNetticketid()+"'";
								Query updatequery = JPA.em().createNativeQuery(TICKET_UPDATE_SQL);
								if(selfsaleticket.getTaketime() != null ){
									updatequery.setParameter("taketime", sdf.parse(selfsaleticket.getTaketime()));
								}
								updatequery.setParameter("updatetime", new Timestamp(updatetime)).executeUpdate();
							}
						}
						if(ts == null){
							ts = alreadyinsertticketsells.get(selfsaleticket.getNetticketid());
						}
						if(selfsaleticket.getReturnway() != null && selfsaleticket.getReturnway().trim().length() > 0
								&& selfsaleticket.getReturnfee() != null ){
							Ticketreturn ticketreturn = existsticketreturns.get(selfsaleticket.getNetticketid());
							if(ticketreturn == null && alreadyinsertticketreturns.get(selfsaleticket.getNetticketid()) == null){
								
								ticketreturn= new Ticketreturn();
								ticketreturn.setId(ts.id);
								ticketreturn.setNetticketid(selfsaleticket.getNetticketid());
								ticketreturn.setDepartorgcode(selfsaleticket.getDepartorgcode());
								ticketreturn.setDepartdate(departdate);
								ticketreturn.setReturnorgcode(selfsaleticket.getReturnorgcode());
								ticketreturn.setTicketno(selfsaleticket.getTicketno());
								ticketreturn.setTicketprice(new BigDecimal(selfsaleticket.getPrice()));
								ticketreturn.setCharges(selfsaleticket.getReturnfee());
								ticketreturn.setReturnamount((selfsaleticket.getPrice())-(selfsaleticket.getReturnfee()));
								ticketreturn.setReturnvoucherno(selfsaleticket.getReturnvoucherno());
								ticketreturn.setReturnname("");
								String returntime = selfsaleticket.getReturntime();
								if(returntime!=null && !"".equals(returntime)){
									ticketreturn.setReturntime(sdf.parse(selfsaleticket.getReturntime()));
								}else{
									ticketreturn.setReturntime(new Date(0));
								}
								ticketreturn.setReturnip("127.0.0.1");
								ticketreturn.setTicketoutlets(selfsaleticket.getTicketoutlets());
								ticketreturn.setReturnway(selfsaleticket.getReturnway());
								ticketreturn.setReturnfeemode(selfsaleticket.getReturnfeemode());
								ticketreturn.setTurnoverdetailid(0);
								Date date=new Date();
								if(returntime!=null && !"".equals(returntime)){
									ticketreturn.setCreatetime(sdf.parse(selfsaleticket.getReturntime()));
								}else{
									ticketreturn.setCreatetime(new Date(0));
								}
								ticketreturn.setCreateby(0);
								ticketreturn.setUpdatetime(date);
								ticketreturn.setUpdateby(0);
								//2018-01-02退票方式
								ticketreturn.setReturntype(selfsaleticket.getReturntype());								
								JPA.em().persist(ticketreturn);
								alreadyinsertticketreturns.put(ticketreturn.getNetticketid(), ticketreturn);
							}else{
								String setReturnvoucherno = "";
								String setReturntime = "";
								if(selfsaleticket.getReturnvoucherno() != null && ((selfsaleticket.getReturnvoucherno().trim().length()) > 0)){
									setReturnvoucherno=" returnvoucherno='"+selfsaleticket.getReturnvoucherno()+"',";
								}
								if(selfsaleticket.getReturntime() != null && selfsaleticket.getReturntime().trim().length()>0){
									setReturntime = "returntime= :returntime,";
								}
								String REFUND_SYN_UPDATE = "update ticketreturn set "+setReturnvoucherno+"returnway='"+selfsaleticket.getReturnway()+"',"
							            +"charges='"+selfsaleticket.getReturnfee()+"',returnfeemode='"+selfsaleticket.getReturnfeemode()+"',"+setReturntime
										+"returnorgcode='"+selfsaleticket.getReturnorgcode()+"' where " +
										" netticketid='"+selfsaleticket.getNetticketid()+"'";
								Query reupdateQuery = JPA.em().createNativeQuery(REFUND_SYN_UPDATE);
								if(selfsaleticket.getReturntime() != null && selfsaleticket.getReturntime().trim().length()>0){
									reupdateQuery.setParameter("returntime", sdf.parse(selfsaleticket.getReturntime()));
								}
								reupdateQuery.executeUpdate();
							} 
							
							//废票不处理，当正常票结算==2017-03-21==
						}else if(selfsaleticket.getTicketstatus() != null && "2".equals(selfsaleticket.getTicketstatus())){
//							if(selfsaleticket.getCanceltime() == null || selfsaleticket.getCanceltime().trim().length() <= 0){
//								//车站实现为处理废票状态
//								log.error(selfsaleticket.getDepartorgcode()+"车站有废票状态的车票，但未处理废票:"+selfsaleticket.getNetticketid()+" 。放弃插入废票记录！！！");
//								continue;
//							}
//							//如果是废票状态，且中心库不存在废票记录，要插入TicketCancel表
//							Ticketcancel ticketcancel = existsticketcancels.get(selfsaleticket.getNetticketid());
//							if(ticketcancel == null && alreadyinsertticketcancels.get(selfsaleticket.getNetticketid()) == null){
//								ticketcancel= new Ticketcancel();
//								ticketcancel.setId(ts.id);
//								ticketcancel.setNetticketid(ts.netticketid);
//								ticketcancel.setDepartorgcode(ts.departorgcode);
//								ticketcancel.setDepartdate(ts.departdate);
//								ticketcancel.setCancelorgcode(selfsaleticket.getCancelorgcode());
//								ticketcancel.setCancelticketoutlets(selfsaleticket.getCancelorgname());
//								ticketcancel.setCancelname(selfsaleticket.getCancelorgname());
//								ticketcancel.setCancelway(selfsaleticket.getCancelway());
//								ticketcancel.setTicketno(selfsaleticket.getTicketno());
//								ticketcancel.setTicketprice(new BigDecimal(selfsaleticket.getPrice()));
//								ticketcancel.setCanceltime(sdf.parse(selfsaleticket.getCanceltime()));
//								ticketcancel.setCancelip("127.0.0.1");
//								ticketcancel.setCancelmode(selfsaleticket.getCancelfeemode());
//								ticketcancel.setTurnoverdetailid(0);
//								Date date=new Date();
//								ticketcancel.setCreatetime(sdf.parse(selfsaleticket.getCanceltime()));
//								ticketcancel.setCreateby(0);
//								ticketcancel.setUpdatetime(date);
//								ticketcancel.setUpdateby(0);
//								JPA.em().persist(ticketcancel);
//								alreadyinsertticketcancels.put(ticketcancel.netticketid, ticketcancel);
//							}
						}
					}catch(Exception e){
						e.printStackTrace();
						log.error("自助售票车票同步失败1", e);
						flag = false;
						if (!JPA.em().getTransaction().isActive()) {//开启事务
							JPA.em().getTransaction().begin();
						}
						JPA.em().getTransaction().rollback();
					}
				}
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			}catch(Exception e){
				log.error("自助售票车票同步失败2", e);
				flag = false;
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
			}
		}catch(Exception e){
			log.error("自助售票车票同步失败3", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.SelfSaleTicket), messages);
			}
		}
		log.info("自助售票车票同步消费结束-------------------------------------------"+sdaf.format(new Date()));
	}

	public void updateexchange(List<String> messages) {
		if(!checkCanConsume(Topic.UpdateExchangefee.name())){
			return;
		}
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
		log.info("updateexchange数据开始消费-----------"+messages.size()+"  条----------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		Integer handleflag = 0;
		try{
			Map<String,ExchangefeeInfo> einfos = new HashMap<String,ExchangefeeInfo>();
			Map<String,Order> eorders = new HashMap<String,Order>();
			StringBuffer sb = new StringBuffer();
			for (String message : messages) {
				List<Map> infos = (ArrayList<Map>) JsonUtil.readJson2Entity(message,ArrayList.class);
				
				for(Map info : infos){
					ExchangefeeInfo ei = einfos.get(info.get("orderno"));
					if( ei != null){
						if(sdf.parse(info.get("updatetime").toString()).after(sdf.parse(ei.updatetime))){
							ei = new ExchangefeeInfo();
							ei.orderno = info.get("orderno").toString();
							ei.updatetime = info.get("updatetime").toString();
							ei.flag = Integer.valueOf(info.get("flag").toString());
							ei.refundexchangefee = Double.valueOf(info.get("refundexchangefee").toString());
							ei.refundmoney = Double.valueOf(info.get("refundmoney").toString());
							ei.refundtime = info.get("refundtime")==null?null:info.get("refundtime").toString();
							einfos.put(info.get("orderno").toString(), ei);
						}
					}else{
						ei = new ExchangefeeInfo();
						ei.orderno = info.get("orderno").toString();
						ei.updatetime = info.get("updatetime").toString();
						ei.flag = Integer.valueOf(info.get("flag").toString());
						ei.refundexchangefee = Double.valueOf(info.get("refundexchangefee").toString());
						ei.refundmoney = Double.valueOf(info.get("refundmoney").toString());
						ei.refundtime = info.get("refundtime")==null?null:info.get("refundtime").toString();
						einfos.put(info.get("orderno").toString(), ei);
					}
				}
			}
			if(einfos.size()<=0){
				return;
			}
			for(String orderno : einfos.keySet()){
				sb.append("'").append(orderno).append("',");
			}
			String ordernos = sb.toString().substring(0, sb.toString().length()-1);
			List<Order> orders = JPA.em().createQuery("select o from Order o where o.orderno in ("+ordernos+")", Order.class)
					.getResultList();
			for(Order o : orders ){
				eorders.put(o.orderno, o);
			}
			if (!JPA.em().getTransaction().isActive()) {//开启事务
				JPA.em().getTransaction().begin();
			}
			for(String orderno : einfos.keySet()){
				Order order = eorders.get(orderno);
				if(order == null){
					log.error( "未找到订单 ："+orderno);
					flag = false;
					continue;
				}
				ExchangefeeInfo info = einfos.get(orderno);
				if(info.flag==0){ // 0-更新orders表交易手续费为0
					Date updatetime = sdf.parse(info.updatetime);
					if(updatetime.after(order.updatetime)){
						JPA.em().createNativeQuery("update orders set exchangefee='0'," +
								"updatetime= :updatetime where" +
								" id='"+order.id+"'").setParameter("updatetime", updatetime).executeUpdate();
					}
				}else{ // 1-更新退款手续费，插入refundexchangefee表
					Refundexchangefee refee = new Refundexchangefee();
					refee.createtime=sdf.parse(info.updatetime);
					refee.orderno = info.orderno;
					refee.refundexchangefee = info.refundexchangefee;
					refee.refundmoney = info.refundmoney;
					refee.refundtime = sdf.parse(info.refundtime);
					JPA.em().persist(refee);
				}
			}
			if (!JPA.em().getTransaction().isActive()) {//开启事务
				JPA.em().getTransaction().begin();
			}
			JPA.em().getTransaction().commit();
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
			JPA.em().clear();
				
		}catch(Exception e){
			log.error( "更新手续费失败", e);
			flag = false;
			if (!JPA.em().getTransaction().isActive()) {//开启事务
				JPA.em().getTransaction().begin();
			}
			JPA.em().getTransaction().rollback();
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.UpdateExchangefee), messages);
			}
		}
		log.info("updateexchange消费结束----------------------------"+sdaf.format(new Date()));
	}
	
	public void ticketsourcessyn(List<String> messages) {
		if(!checkCanConsume(Topic.Ticketsources.name())){
			return;
		}
		log.info("车票数据来源开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			Map<String,Stationselldetail> stationsells = new HashMap<String,Stationselldetail>();
			Map<String,Stationreturndetail> stationreturns = new HashMap<String,Stationreturndetail>();
			Map<String,Stationcanceldetail> stationcancels = new HashMap<String,Stationcanceldetail>();
			List<String> sellpklist = new ArrayList<String>();
			List<String> returnpklist = new ArrayList<String>();
			List<String> cancelpklist = new ArrayList<String>();
			Map<String,Tkusedetail> tkuses = new HashMap<String,Tkusedetail>();
			StringBuilder sellpksb = new StringBuilder();
			StringBuilder returnpksb = new StringBuilder();
			StringBuilder cancelpksb = new StringBuilder();
			StringBuilder tkpksb = new StringBuilder();
			for(String message : messages){
				JsonNode jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				Ticketsources ts = (Ticketsources) JsonUtil.readJson2Entity(content, Ticketsources.class);
				if( null != ts.getTicketstatus() && ts.getTicketstatus().trim().length()>0){
					sellpksb.append("'").append(ts.getStationsellpk()).append("',");
					if("1".equals(ts.getTicketstatus().trim())){
						returnpksb.append("'").append(ts.getStationreturnpk()).append("',");
					}else if("2".equals(ts.getTicketstatus().trim())){
						cancelpksb.append("'").append(ts.getStationcancelpk()).append("',");
					}
				}
				if(ts.getIsconsumesellticket()!=null && "1".equals(ts.getIsconsumesellticket().trim())){
					tkpksb.append("'").append(ts.getStationsellpk()).append("',");
				}
				if(ts.getIsconsumesellticket()!=null && "1".equals(ts.getIsconsumereturnticket().trim())){
					tkpksb.append("'").append(ts.getStationreturnpk()).append("',");
				}
			}
			log.info("车票数据来源开始查询售票数据----------------------"+sdaf.format(new Date()));
			if(sellpksb.toString().trim().length() > 0){
				String sellpks = sellpksb.toString().substring(0, sellpksb.toString().length()-1);
				List<Stationselldetail> stationsellresult = JPA.em().createQuery("select o from Stationselldetail o where o.stationpk in ("+sellpks+")", Stationselldetail.class)
						.getResultList();
				for(Stationselldetail sell : stationsellresult){
					stationsells.put(sell.stationpk, sell);
				}
			}
			log.info("车票数据来源查询售票数据结束并开始查询退票数据----------------------"+sdaf.format(new Date()));
			if(returnpksb.toString().trim().length() > 0){
				String returnpks = returnpksb.toString().substring(0, returnpksb.toString().length()-1);
				List<Stationreturndetail> stationreturnresult = JPA.em().createQuery("select o from Stationreturndetail o where o.stationpk in ("+returnpks+")", Stationreturndetail.class)
						.getResultList();
				for(Stationreturndetail srt : stationreturnresult){
					stationreturns.put(srt.stationpk, srt);
				}
			}
			log.info("车票数据来源查询退票数据结束并开始查询废票数据----------------------"+sdaf.format(new Date()));
			if(cancelpksb.toString().trim().length() > 0){
				String cancelpks = cancelpksb.toString().substring(0, cancelpksb.toString().length()-1);
				List<Stationcanceldetail> stationcancelresult = JPA.em().createQuery("select o from Stationcanceldetail o where o.stationpk in ("+cancelpks+")", Stationcanceldetail.class)
						.getResultList();
				for(Stationcanceldetail srt : stationcancelresult){
					stationcancels.put(srt.stationpk, srt);
				}
			}
			log.info("车票数据来源查询废票数据结束并开始查询票证使用记录----------------------"+sdaf.format(new Date()));
			if(tkpksb.toString().trim().length() > 0){
				String tkpks = tkpksb.toString().substring(0, tkpksb.toString().length()-1);
				List<Tkusedetail> tkuseresult = JPA.em().createQuery("select o from Tkusedetail o where o.stationpk in ("+tkpks+")", Tkusedetail.class)
						.getResultList();
				for(Tkusedetail srt : tkuseresult){
					tkuses.put(srt.stationpk, srt);
				}
			}
			log.info("车票数据来源查询票证使用记录结束----------------------"+sdaf.format(new Date()));
			log.info("车票数据来源开始插入数据库----------------------"+sdaf.format(new Date()));
			for(String message : messages){
				try{
					if (!JPA.em().getTransaction().isActive()) {//开启事务
						JPA.em().getTransaction().begin();
					}
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					Ticketsources ts = (Ticketsources) JsonUtil.readJson2Entity(content, Ticketsources.class);
					if(ts.getDepartdate()==null||"".equals(ts.getDepartdate())){
						continue;
					}
					Date updatetime = datetimeformat.parse(ts.getUpdatetime());
					if( null != ts.getTicketstatus() && ts.getTicketstatus().trim().length()>0){
						Stationselldetail sell = stationsells.get(ts.getStationsellpk());
						if(sell == null && !sellpklist.contains(ts.getStationsellpk())){
							sell = new Stationselldetail();
							sell.stationpk = ts.getStationsellpk();
							sell.ticketno = ts.getTicketno();
							sell.departorgcode = ts.getDepartorgcode();
							sell.reachstationcode = ts.getReachstationcode();
							sell.schedulecode = ts.getSchedulecode();
							sell.departdate = dateformat.parse(ts.getDepartdate());
							sell.departtime = ts.getDeparttime();
							sell.seatno = ts.getSeatno();
							sell.price = ts.getPrice();
							sell.distance = ts.getDistance();
							sell.selltime = datetimeformat.parse(ts.getSelltime());
							sell.sellorgcode = ts.getSellorgcode();
							sell.selltype = ts.getSelltype();
							sell.ticketstatus = ts.getTicketstatus();
							sell.takestatus = ts.getTakestatus();
							if(sell.takestatus !=null && "2".equals(sell.takestatus.trim())&&ts.getTaketime()!=null){
								sell.taketime = datetimeformat.parse(ts.getTaketime());
							}
							sell.remark = ts.getRemark();
							sell.updateby = "0";
							sell.updatetime = updatetime;
							sell.createtime = new Date();
							sell.createby = "0";
							sell.routecode = ts.getRoutecode();//	线路编码
							sell.routename = ts.getRoutename();//	线路名称
							sell.endstationcode = ts.getEndstationcode();//	终到站编码
							sell.endstationname = ts.getEndstationname();//	终到站名称
							sell.vehicleno = ts.getVehicleno();//	车牌号
							sell.islinework = ts.getIslinework();//	是否流水班
							sell.reachstationname = ts.getReachstationname();//	到站名称
							JPA.em().persist(sell);
							sellpklist.add(sell.stationpk);
						}else if(sell!=null){
							if(sell.updatetime.before(updatetime)){
								if(!sell.ticketstatus.trim().equals(ts.getTicketstatus().trim())){
									sell.ticketstatus = ts.getTicketstatus();
								}
								if((!"2".equals(sell.takestatus.trim())) && "2".equals(ts.getTakestatus().trim())){
									sell.takestatus = ts.getTakestatus();
									sell.taketime = datetimeformat.parse(ts.getTaketime());
								}
								sell.updatetime = updatetime;
								JPA.em().merge(sell);
//								JPA.em().persist(sell);
							}
						}
						if("1".equals(ts.getTicketstatus().trim())){
							Stationreturndetail sreturn = stationreturns.get(ts.getStationreturnpk());
							if(sreturn == null && !returnpklist.contains(ts.getStationreturnpk())){
								sreturn = new Stationreturndetail();
								sreturn.ticketselldetailid = sell.id;   
								sreturn.stationpk = ts.getStationreturnpk();             
								sreturn.ticketno = ts.getReturnbillno();           
								sreturn.departorgcode = sell.departorgcode;      
								sreturn.reachstationcode = sell.reachstationcode;   
								sreturn.schedulecode = sell.schedulecode;
								sreturn.departdate = sell.departdate;           
								sreturn.departtime = sell.departtime;         
								sreturn.seatno = sell.seatno;             
								sreturn.price = sell.price;              
								sreturn.distance = sell.distance;
								sreturn.returntime = ts.getReturntime()!=null?datetimeformat.parse(ts.getReturntime()):null;           
								sreturn.returnorgcode = ts.getReturnorgcode();      
								sreturn.returntype = ts.getReturntype();         
								sreturn.returnfee = ts.getReturnfee();          
								sreturn.remark = "";             
								sreturn.updateby = "0";
								sreturn.updatetime = updatetime;
								sreturn.createtime = new Date();            
								sreturn.createby = "0"; 
								sreturn.routecode = ts.getRoutecode();//	线路编码
								sreturn.routename = ts.getRoutename();//	线路名称
								sreturn.islinework = ts.getIslinework();//	是否流水班
								JPA.em().persist(sreturn);
								returnpklist.add(ts.getStationreturnpk());
							}
						}else if("2".equals(ts.getTicketstatus().trim())){
							Stationcanceldetail cancel = stationcancels.get(ts.getStationcancelpk());
							if(cancel == null && !cancelpklist.contains(ts.getStationcancelpk())){
								cancel = new Stationcanceldetail();
								cancel.ticketselldetailid = sell.id;   
								cancel.stationpk = ts.getStationreturnpk();             
								cancel.ticketno = ts.getReturnbillno();           
								cancel.departorgcode = sell.departorgcode;      
								cancel.reachstationcode = sell.reachstationcode;   
								cancel.schedulecode = sell.schedulecode;
								cancel.departdate = sell.departdate;           
								cancel.departtime = sell.departtime;         
								cancel.seatno = sell.seatno;             
								cancel.price = sell.price;              
								cancel.distance = sell.distance;
								cancel.canceltime = ts.getCanceltime()!=null?datetimeformat.parse(ts.getCanceltime()):null;  
								cancel.cancelorgcode = ts.getCancelorgcode();      
								cancel.canceltype = ts.getCanceltype();        
								cancel.remark = "";             
								cancel.updateby = "0";
								cancel.updatetime = updatetime;
								cancel.createtime = new Date();            
								cancel.createby = "0"; 
								JPA.em().persist(cancel);
								cancelpklist.add(ts.getStationcancelpk());
							}
						}
					}
					if(ts.getIsconsumesellticket()!=null && "1".equals(ts.getIsconsumesellticket().trim())){
						Tkusedetail tu = tkuses.get(ts.getStationsellpk());
						if(tu == null){
							tu = new Tkusedetail();
							tu.receiveorgcode = ts.getReceiveorgcode();
							tu.username = ts.getSellusename();
							tu.typeid = gettktypeid();
							tu.stationpk = ts.getStationsellpk();
							tu.billno = ts.getSellbillno();
							tu.usetime = datetimeformat.parse(ts.getSellusetime());
							tu.usetype = "0";
							tu.price = ts.getPrice();
							tu.usedescription = ts.getSellusedescription();
							tu.ticketoutletsname = ts.getSellticketoutletsname();
							tu.seller = ts.getSeller();
							tu.departorgcode = ts.getDepartorgcode();
							tu.schedulecode = ts.getSchedulecode();
							tu.departdate = dateformat.parse(ts.getDepartdate());
							tu.departtime = ts.getDeparttime();
							tu.seatno = ts.getSeatno();
							tu.selltype = ts.getSelltype();
							tu.createtime = new Date();
							tu.createby = "0";
							tu.updatetime = updatetime;
							tu.updateby = "0";
							JPA.em().persist(tu);
						}
					}
					if(ts.getIsconsumesellticket()!=null && "1".equals(ts.getIsconsumereturnticket().trim())){
						Tkusedetail tu = tkuses.get(ts.getStationreturnpk());
						if(tu == null){
							tu = new Tkusedetail();
							tu.receiveorgcode = ts.getReceiveorgcode();
							tu.username = ts.getReturnusename();
							tu.typeid = gettktypeid();
							tu.stationpk = ts.getStationreturnpk();
							tu.billno = ts.getReturnbillno();
							tu.usetime = datetimeformat.parse(ts.getReturnusetime());
							tu.usetype = "1";
							tu.price = ts.getPrice();
							tu.usedescription = ts.getReturnusedescription();
							tu.ticketoutletsname = ts.getReturnticketoutletsname();
							tu.seller = ts.getSeller();
							tu.departorgcode = ts.getDepartorgcode();
							tu.schedulecode = ts.getSchedulecode();
							tu.departdate = dateformat.parse(ts.getDepartdate());
							tu.departtime = ts.getDeparttime();
							tu.seatno = ts.getSeatno();
							tu.selltype = ts.getSelltype();
							tu.createtime = new Date();
							tu.createby = "0";
							tu.updatetime = updatetime;
							tu.updateby = "0";
							JPA.em().persist(tu);
						}
					}
				}catch(Exception e){
									e.printStackTrace();

	Logger.log4j.error("消费车票数据来源消息失败1", e);
					flag = false;
					if (!JPA.em().getTransaction().isActive()) {//开启事务
						JPA.em().getTransaction().begin();
					}
					JPA.em().getTransaction().rollback();
				}
			}
			if (!JPA.em().getTransaction().isActive()) {//开启事务
				JPA.em().getTransaction().begin();
			}
			JPA.em().getTransaction().commit();
		}catch(Exception e){
			log.error("消费车票数据来源消息失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Ticketsources), messages);
			}
			JPA.em().clear();
		}
		log.info("车票数据来源消费结束---------------------------"+sdaf.format(new Date()));
	}
	

	public void stationstaffsyn(List<String> messages) {
		if(!checkCanConsume(Topic.Stationstaff.name())){
			return;
		}
		log.info("车站从业人员信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			Map<Long,Staff> staffmap = new HashMap<Long,Staff>();
			StringBuilder hashidsb = new StringBuilder();
			for(String message : messages){
				JsonNode jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				Stationstaff ss = (Stationstaff) JsonUtil.readJson2Entity(content, Stationstaff.class);
				if(ss!=null && (ss.getHashid()==null || ss.getHashid()==0)){
					continue;
				}
				Date updatetime = datetimeformat.parse(ss.getUpdatetime());
				Staff staff = staffmap.get(ss.getHashid());
				if(staff == null || staff.updatetime.before(updatetime)){
					Date birthday = null;
					try{
						birthday = dateformat.parse(ss.getBirthday());
					}catch(Exception e){
					}
					staff = new Staff();
					staff.orgcode = ss.getOrgcode();
					staff.unitcode = ss.getUnitcode();
					staff.staffproperty = ss.getStaffproperty();
					staff.name = ss.getName();
					staff.sex = ss.getSex();
					staff.birthday =  birthday;
					staff.posttype = ss.getPosttype();
					staff.certificatesno = ss.getCertificatesno();
					staff.isactive = "1".equals(ss.getIsactive().trim());
					staff.remark = ss.getRemark();
					staff.updateby = "0";
					staff.updatetime = updatetime;
					staff.createtime = new Date();
					staff.createby = "0";
					staff.hashid=ss.getHashid();
					staffmap.put(ss.getHashid(), staff);
				}
				hashidsb.append("'").append(ss.getHashid()).append("',");
			}
			if(!"".equals(hashidsb.toString())){
				try {
					String hashids = hashidsb.toString().substring(0, hashidsb.toString().length()-1);
					if (!JPA.em().getTransaction().isActive()) {//开启事务
						JPA.em().getTransaction().begin();
					}
					String DEL_STAFF_SQL = "delete from staff  " +
							" where hashid in ("+hashids+")";
					JPA.em().createNativeQuery(DEL_STAFF_SQL).executeUpdate();
					for (Staff staff : staffmap.values()) {
						JPA.em().persist(staff);
					}
					JPA.em().getTransaction().commit();
					
				} catch (Exception e) {
					if (!JPA.em().getTransaction().isActive()) {//开启事务
						JPA.em().getTransaction().begin();
					}
					JPA.em().getTransaction().rollback();
					log.error("消费车站从业人员信息失败2", e);
					flag = false;
				}
			}
		}catch(Exception e){
			log.error("消费车站从业人员信息失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Stationstaff), messages);
			}
			JPA.em().clear();
		}
		log.info("车站从业人员信息消费结束---------------------------"+sdaf.format(new Date()));
	}
	
	
	public void outletsinfosyn(List<String> messages) {
		if(!checkCanConsume(Topic.Outletsinfo.name())){
			return;
		}
		log.info("代售点信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			Map<Long,Outlets> outletsmap = new HashMap<Long,Outlets>();
			StringBuilder hashidsb = new StringBuilder();
			for(String message : messages){
				JsonNode jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				Outletsinfo ss = (Outletsinfo) JsonUtil.readJson2Entity(content, Outletsinfo.class);
				Date updatetime = datetimeformat.parse(ss.getUpdatetime());
				Outlets oi = outletsmap.get(ss.getHashid());
				if(oi == null || oi.updatetime.before(updatetime)){
					oi = new Outlets();
					oi.orgcode = ss.getOrgcode();
					oi.nature = ss.getNature();
					oi.adress = ss.getAdress();
					oi.name = ss.getName();
					oi.isactive = ss.getIsactive();
					oi.head = ss.getHead();
					oi.linkname = ss.getLinkname();
					oi.linkphone = ss.getLinkphone();
					oi.businesstime = ss.getBusinesstime();
					oi.remark = ss.getRemark();
					oi.updateby = "0";
					oi.updatetime = updatetime;
					oi.createtime = new Date();
					oi.createby = "0";
					oi.hashid = ss.getHashid();
					outletsmap.put(ss.getHashid(), oi);
				}
				hashidsb.append("'").append(ss.getHashid()).append("',");
			}
			try {
				String hashids = hashidsb.toString().substring(0, hashidsb.toString().length()-1);
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				String DEL_OUTLETS_SQL = "delete from outlets  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_OUTLETS_SQL).executeUpdate();
				for (Outlets outlets : outletsmap.values()) {
					JPA.em().persist(outlets);
				}
				JPA.em().getTransaction().commit();
				
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("消费代售点信息失败2", e);
				flag = false;
			}
			
		}catch(Exception e){
			log.error("消费代售点信息失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Outletsinfo), messages);
			}
			JPA.em().clear();
		}
		log.info("代售点信息消费结束---------------------------"+sdaf.format(new Date()));
	}
	
	public void stationbehindschedulesyn(List<String> messages) {
		if(!checkCanConsume(Topic.Stationbehindschedule.name())){
			return;
		}
		log.info("车站脱班信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			Map<Long,Behindschedule> behindmap = new HashMap<Long,Behindschedule>();
			StringBuilder hashidsb = new StringBuilder();
			for(String message : messages){
				JsonNode jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				Stationbehindschedule ss = (Stationbehindschedule) JsonUtil.readJson2Entity(content, Stationbehindschedule.class);
				Date updatetime = datetimeformat.parse(ss.getUpdatetime());
				Behindschedule oi = behindmap.get(ss.getHashid());
				if(oi == null || oi.updatetime.before(updatetime)){
					oi = new Behindschedule();
					oi.orgcode = ss.getOrgcode();             
					oi.schedulecode = ss.getSchedulecode(); 
					oi.departdate = dateformat.parse(ss.getDepartdate());          
					oi.departtime = ss.getDeparttime();          
					oi.unitcode = ss.getUnitcode();            
					oi.behindnum = ss.getBehindnum();           
					oi.remark = ss.getRemark();              
					oi.updateby = "0"; 
					oi.updatetime = updatetime; 
					oi.createtime = new Date();          
					oi.createby = "0";  
					oi.hashid = ss.getHashid();
					behindmap.put(ss.getHashid(), oi);
				}
				hashidsb.append("'").append(ss.getHashid()).append("',");
			}
			try {
				String hashids = hashidsb.toString().substring(0, hashidsb.toString().length()-1);
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				String DEL_OUTLETS_SQL = "delete from behindschedule  " +
						" where hashid in ("+hashids+")";
				JPA.em().createNativeQuery(DEL_OUTLETS_SQL).executeUpdate();
				for (Behindschedule behind : behindmap.values()) {
					JPA.em().persist(behind);
				}
				JPA.em().getTransaction().commit();
				
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("消费车站脱班信息失败2", e);
				flag = false;
			}
			
		}catch(Exception e){
			log.error("消费车站脱班信息失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Stationbehindschedule), messages);
			}
			JPA.em().clear();
		}
		log.info("车站脱班信息消费结束---------------------------"+sdaf.format(new Date()));
	}
	
	/**
	 * 车站发送旅客信息消费
	 * @param messages
	 */
	public void stationsendssync(List<String> messages) {
		if(!checkCanConsume(Topic.StationSends.name())){
			return;
		}
		log.info("车站发送旅客信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		List<String> orglist = new ArrayList<String>();
		try{
			List<Stationsend> stationsendsmap = new ArrayList<Stationsend>();
			String departdate = "";
			String orgcode = "";
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					Stationsend stationend = (Stationsend) JsonUtil.readJson2Entity(content, Stationsend.class);
					stationsendsmap.add(stationend);
					
					if(departdate==null || departdate.equals("")){
						departdate = stationend.getDepartdate();
					}
					if(departdate==null || orgcode.equals("")){
						orgcode = stationend.getOrgcode();
					}
					if(!orglist.contains(orgcode+departdate)){
						orglist.add(orgcode+departdate);
					}
				} catch (Exception e) {
					flag = false;
				}
			}		
			Map<String,String> redismap = RedisUtil.get(RedisUtil.RedisDBApp.stationsendmsgtime,orglist);
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				for (Stationsend stationend : stationsendsmap) {
					Long msgtime = 0L;
					if(redismap.get(stationend.getOrgcode()+stationend.getDepartdate()) != null){
						msgtime = Long.parseLong(redismap.get(stationend.getOrgcode()+stationend.getDepartdate()));
					}else{
						String sql = "select ifnull(max(msgtime),0) from departpassenger where orgcode=:orgcode and departdate=:departdate";
						Long msgtime_db = null;
						try {
							Query query = JPA.em().createNativeQuery(sql);
							query.setParameter("orgcode", stationend.getOrgcode());
							query.setParameter("departdate", DateUtils.StrToDate(stationend.getDepartdate()));
							List<Object> list = query.getResultList();
							
							if(list!=null && list.size()>0){
								msgtime_db = Long.valueOf(list.get(0).toString());
							}
						} catch (Exception e) {
							Logger.log4j.info("查询最大消息发送日期出错，orgcode='"+stationend.getOrgcode()+"' and departdate='"+stationend.getDepartdate()+"'",e);
							msgtime_db = null;
						}
						if(msgtime_db!=null){
							msgtime = msgtime_db;
						}
					}
					deleteDepartPassengerByDepartdate(stationend.getDepartdate(), stationend.getOrgcode(),msgtime);
					if(msgtime <= stationend.getMsgtime()){
						stationend.setCreatetime(new Date());
						JPA.em().persist(stationend);
					}
					if(msgtime < stationend.getMsgtime()){
						redismap.put(stationend.getOrgcode()+stationend.getDepartdate(), stationend.getMsgtime().toString());
					}
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
				//批量插入redis
				RedisUtil.set(RedisDBApp.stationsendmsgtime, redismap);
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化车站发送旅客信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化车站发送旅客信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.StationSends), messages);
			}
		}
		log.info("stationsend消费结束-------------------------"+sdaf.format(new Date()));
	}
	
	public void payrecordsync(List<String> messages) {
		if(!checkCanConsume(Topic.PayRecords.name())){
			return;
		}
		log.info("自助机支付信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		List<Long> hashids = new ArrayList<Long>();
		try{
			List<PayRecord> payrecordsmap = new ArrayList<PayRecord>();
			
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					PayRecord payrecord = (PayRecord) JsonUtil.readJson2Entity(content, PayRecord.class);
					payrecordsmap.add(payrecord);
					hashids.add(payrecord.getHashid());
				} catch (Exception e) {
					flag = false;
				}
			}		
			try {
				Map<Long,PayRecord> alreadyIns = new HashMap<Long, PayRecord>();
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				String DEL_SCHEDULEPRICE_SQL = "delete from payrecord  " +
						" where hashid in (:hashids)";
				JPA.em().createNativeQuery(DEL_SCHEDULEPRICE_SQL)
						.setParameter("hashids", hashids).executeUpdate();
				JPA.em().getTransaction().commit();
				
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				Date createtime = new Date();
				for (PayRecord payrecord : payrecordsmap) {
					if(alreadyIns.get(payrecord.getHashid()) != null){
						continue;
					}else{
						alreadyIns.put(payrecord.getHashid(), payrecord);
					}
					payrecord.setCreatetime(createtime);
					JPA.em().persist(payrecord);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化自助机支付信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化自助机支付信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.PayRecords), messages);
			}
		}
		log.info("payrecords消费结束-------------------------"+sdaf.format(new Date()));
	}

	public void lossticketssync(List<String> messages) {
		if(!checkCanConsume(Topic.LossTickets.name())){
			return;
		}
		log.info("错误车票票证信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		List<Long> hashids = new ArrayList<Long>();
		try{
			List<Ticketlossdetail> payrecordsmap = new ArrayList<Ticketlossdetail>();
			
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					Ticketlossdetail payrecord = (Ticketlossdetail) JsonUtil.readJson2Entity(content, Ticketlossdetail.class);
					payrecordsmap.add(payrecord);
					hashids.add(payrecord.getHashid());
				} catch (Exception e) {
					flag = false;
				}
			}		
			try {
				Map<Long,Ticketlossdetail> alreadyIns = new HashMap<Long, Ticketlossdetail>();
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				String DEL_SCHEDULEPRICE_SQL = "delete from tk_loss_detail  " +
						" where hashid in (:hashids)";
				JPA.em().createNativeQuery(DEL_SCHEDULEPRICE_SQL)
						.setParameter("hashids", hashids).executeUpdate();
				JPA.em().getTransaction().commit();
				
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				Date createtime = new Date();
				for (Ticketlossdetail payrecord : payrecordsmap) {
					if(alreadyIns.get(payrecord.getHashid()) != null){
						continue;
					}else{
						alreadyIns.put(payrecord.getHashid(), payrecord);
					}
					payrecord.setCreatetime(datetimeformat.format(createtime));
					JPA.em().persist(payrecord);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化错误车票票证信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化错误车票票证信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.LossTickets), messages);
			}
		}
		log.info("LossTickets消费结束-------------------------"+sdaf.format(new Date()));
	}

	public void stationinsurancesync(List<String> messages) {
		if(!checkCanConsume(Topic.Stationinsurance.name())){
			return;
		}
		log.info("车站保险信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			Map<Long,StationInsurance> stationsendsmap = new HashMap<Long,StationInsurance>();
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					StationInsurance stationinr = (StationInsurance) JsonUtil.readJson2Entity(content, StationInsurance.class);
					stationsendsmap.put(stationinr.getHashid(),stationinr);
				} catch (Exception e) {
					flag = false;
				}
			}		
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				deleteStationInsureByHashids(stationsendsmap.keySet());
				for (StationInsurance stationinsure : stationsendsmap.values()) {
					stationinsure.setCreatetime(new Date());
					JPA.em().persist(stationinsure);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化车站保险信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化车站保险信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Stationinsurance), messages);
			}
		}
		log.info("Stationinsurance消费结束-------------------------"+sdaf.format(new Date()));
	}
	
	private void deleteStationInsureByHashids(Set<Long> hashids) {
		String sql = "delete from stationinsurance where hashid in(:hashids)";
		Query queryOrganization = JPA.em().createNativeQuery(sql);
		queryOrganization.setParameter("hashids", hashids);
		try {
			queryOrganization.executeUpdate();
		} catch (Exception e) {
			Logger.log4j.error("删除stationinsurance失败hashids="+hashids, e);
		}
	}

	public void fullreturnschedulepricesync(List<String> messages) {
		if(!checkCanConsume(Topic.FullReturnSchedules.name())){
			return;
		}
		log.info("可全额退款班次信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		List<Long> hashids = new ArrayList<Long>();
		try{
			List<FullReturnSchedulePrice> payrecordsmap = new ArrayList<FullReturnSchedulePrice>();
			
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					FullReturnSchedulePrice payrecord = (FullReturnSchedulePrice) JsonUtil.readJson2Entity(content, FullReturnSchedulePrice.class);
					payrecordsmap.add(payrecord);
					hashids.add(payrecord.getHashid());
				} catch (Exception e) {
					flag = false;
				}
			}		
			try {
				Map<Long,FullReturnSchedulePrice> alreadyIns = new HashMap<Long, FullReturnSchedulePrice>();
//				if (!JPA.em().getTransaction().isActive()) {//开启事务
//					JPA.em().getTransaction().begin();              
//				}
//				String DEL_SCHEDULEPRICE_SQL = "delete from fullreturnscheduleprice  " +
//						" where hashid in (:hashids)";
//				JPA.em().createNativeQuery(DEL_SCHEDULEPRICE_SQL)
//						.setParameter("hashids", hashids).executeUpdate();
//				JPA.em().getTransaction().commit();
				
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				Date createtime = new Date();
				for (FullReturnSchedulePrice payrecord : payrecordsmap) {
					if(alreadyIns.get(payrecord.getHashid()) != null || checkIsExist(payrecord.getHashid())){
						continue;
					}else{
						alreadyIns.put(payrecord.getHashid(), payrecord);
					}
					payrecord.setCreatetime(createtime);
					payrecord.setUpdatetime(createtime);
					JPA.em().persist(payrecord);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化可全额退款班次信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化可全额退款班次信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.FullReturnSchedules), messages);
			}
		}
		log.info("FullReturnSchedules消费结束-------------------------"+sdaf.format(new Date()));
	}

	private boolean checkIsExist(Long hashid) {
		List<FullReturnSchedulePrice> prices = FullReturnSchedulePrice.find("hashid=?", hashid).fetch();
		if(prices.size()==0){
			return false;
		}else{
			return true;
		}
	}

	private void deleteDepartPassengerByDepartdate(String departdate,
			String orgcode, Long msgtime) {
		String sql = "delete from departpassenger where orgcode=:orgcode and departdate=:departdate and msgtime<:msgtime";
		Query queryOrganization = JPA.em().createNativeQuery(sql);
		queryOrganization.setParameter("orgcode", orgcode);
		queryOrganization.setParameter("departdate", DateUtils.StrToDate(departdate));
		queryOrganization.setParameter("msgtime", msgtime);
		try {
			queryOrganization.executeUpdate();
		} catch (Exception e) {
			Logger.log4j.error("删除departpassenger失败orgcode="+orgcode+",departdate="+departdate+"msgtime<"+msgtime, e);
		}
	}

	/*
	 * 更新msgsendlog的消费状态
	 */
	private void finishMsg(String msghashids ,Date dealtime){
		EntityManager em = JPA.newEntityManager();
		try {
			
			EntityTransaction transaction = em.getTransaction();
			transaction.begin();
			String FINISH_MSGSENDLOG = "update msgsendlog set dealed = :dealed,"
					+ " dealtime = :dealtime where msghashid in ("+msghashids+")";
			Query finishQuery = em.createNativeQuery(FINISH_MSGSENDLOG);
			finishQuery.setParameter("dealtime", new Timestamp(dealtime.getTime()));
			finishQuery.setParameter("dealed", true);
			finishQuery.executeUpdate();
			transaction.commit();
		} catch (Exception e) {
			em.getTransaction().rollback();
			log.error("更新消息消费状态失败", e);
		}finally{
			if(em.getTransaction().isActive()){
				em.getTransaction().commit();
			}
			if(em!=null){
				if(em.isOpen()){
					em.close();
				}
			}
		}
	}
	
	private String getSex(String certificateno){
		if(certificateno != null && certificateno.trim().length() == 18){
			if(Integer.valueOf(certificateno.substring(16, 17))%2==1){
				return "男";
			}else{
				return "女";
			}
		}
		return "";
	}
	private Long gettktypeid(){
//		if(tktypeid == null){
//			Object o  = JPA.em().createNativeQuery("select id from tk_type where code = 'Ticket'").getSingleResult();
//			tktypeid = Long.parseLong(o.toString());
//		}
		//因票证系统影响，不用这个字段
		return 1L;
	}
	
	
	/**
	 * 中心kafka消费机构表
	 * @param messages
	 */
	public void netorgtoKafka(List<String> messages) {
		if(!checkCanConsume(Topic.Netcentreorg.name())){
			return;
		}
		log.info("中心机构表信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			Map<Long,NetOrganization> organizationsmap = new HashMap<Long,NetOrganization>();
			for (String message : messages) {
				try {
					
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					NetOrganization netorganization = (NetOrganization) JsonUtil.readJson2Entity(content, NetOrganization.class);
					organizationsmap.put(netorganization.getId(),netorganization);
				} catch (Exception e) {
					flag = false;
				}
			}		
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				deleteNetorgtoKafka(organizationsmap.keySet());
				for (NetOrganization netorganization : organizationsmap.values()) {
					netorganization.setCreatetime(new Date());
					JPA.em().persist(netorganization);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化中心机构表信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化中心机构表信息信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.Netcentreorg), messages);
			}
		}
		log.info("netorgtoKafka消费结束-------------------------"+sdaf.format(new Date()));
	}
	
	private void deleteNetorgtoKafka(Set<Long> ids) {
		String sql = "delete from netorganization where id in(:id)";
		Query queryOrganization = JPA.em().createNativeQuery(sql);
		queryOrganization.setParameter("id", ids);
		try {
			queryOrganization.executeUpdate();
		} catch (Exception e) {
			Logger.log4j.error("删除netorganization表中id="+ids+"失败", e);
		}
	}
	
	public void stationviewgps(List<String> messages) {
		if(!checkCanConsume(Topic.StationViewgps.name())){
			return;
		}
		log.info("车站GPS车辆监管信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		List<String> orglist = new ArrayList<String>();
		try{
			Map<Long,Stationviewgps> stationviewgpsmap = new HashMap<Long,Stationviewgps>();
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					Stationviewgps stationviewgps = (Stationviewgps) JsonUtil.readJson2Entity(content, Stationviewgps.class);
					stationviewgpsmap.put(stationviewgps.getId(),stationviewgps);
				} catch (Exception e) {
					flag = false;
				}
			}		
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				deleteStationviewgps(stationviewgpsmap.keySet());
				for (Stationviewgps stationviewgps : stationviewgpsmap.values()) {
					
					JPA.em().persist(stationviewgps);
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化车站GPS车辆监管信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化车站GPS车辆监管信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.StationViewgps), messages);
			}
		}
		log.info("stationviewgps消费结束-------------------------"+sdaf.format(new Date()));
	}
	
	private void deleteStationviewgps(Set<Long> ids) {
		String sql="delete from stationviewgps where id in(:id)";
		Query deleteQuery = JPA.em().createNativeQuery(sql);
		deleteQuery.setParameter("id", ids);
		try {
			deleteQuery.executeUpdate();
		} catch (Exception e) {
			Logger.log4j.error("删除stationviewgps表中id="+ids+"失败", e);
		}
	}

	public static void main(String[] args) {
		System.out.println(FNVHash.hash64("450700WMKYZX" +"#"+ "粤Cl1802"));
	}

	/**
	 * @param messages
	 */
	public void qhstationdatasyn(List<String> messages) {
		if(!checkCanConsume(Topic.QHStationData.name())){
			return;
		}
		log.info("青海车站发班运量开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
//		List<String> orglist = new ArrayList<String>();
		try{
			Map<Long,Qhstationdata> stationviewgpsmap = new HashMap<Long,Qhstationdata>();
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					Qhstationdata stationviewgps = (Qhstationdata) JsonUtil.readJson2Entity(content, Qhstationdata.class);
					stationviewgpsmap.put(stationviewgps.hashid,stationviewgps);
				} catch (Exception e) {
					flag = false;
				}
			}		
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
//				deleteStationviewgps(stationviewgpsmap.keySet());
				Qhstationdata dbdata = null;
				for (Qhstationdata stationviewgps : stationviewgpsmap.values()) {
					dbdata = Qhstationdata.find("hashid=?", stationviewgps.hashid).first();
					if(dbdata == null){
						JPA.em().persist(stationviewgps);
					}
				}
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化青海车站发班运量数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化青海车站发班运量数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.StationViewgps), messages);
			}
		}
		log.info("青海车站发班运量消费结束-------------------------"+sdaf.format(new Date()));
	}
	public void insurancesync(List<String> messages){
		if(!checkCanConsume(Topic.insurance.name())){
			return;
		}
//		log.info("车站保险信息开始消费---------"+messages.size()+"  条-------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka
		try{
			Map<Long,InterfaceInsurance> stationsendsmap = new HashMap<Long,InterfaceInsurance>();
			StringBuffer netticketids = new StringBuffer();
			String departorgcode = "";
			for (String message : messages) {
				try {
					JsonNode jnode = JsonUtil.getJsonNode(message);
					String content = jnode.get("content").toString();
					InterfaceInsurance insuranceinr = (InterfaceInsurance) JsonUtil.readJson2Entity(content, InterfaceInsurance.class);
					stationsendsmap.put(insuranceinr.hasid,insuranceinr);
//					log.info("hashids-------------------------"+insuranceinr.hasid);
					departorgcode = insuranceinr.departorgcode;
					netticketids.append("'").append(insuranceinr.netticketid).append("'").append(",");
				} catch (Exception e) {
					flag = false;
				}
			}		
			String hashids = netticketids.toString();
//			log.info("hashids-------------------------"+hashids);
			hashids = hashids.substring(0, hashids.length()-1);
			try {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();              
				}
				List<Insurance> insurances = JPA.em().createQuery("select i from Insurance i where  i.netticketid in ("+hashids+")", Insurance.class).getResultList();
//				log.info("insurances数量-------------------------"+insurances.size());
				for (Insurance insurance : insurances) {
					for (InterfaceInsurance ininsure : stationsendsmap.values()) {
						if (insurance.netticketid.equals(ininsure.netticketid)) {
							JPA.em().persist(insurance);
							insurance.updatetime = new Date();
							insurance.insurancestatus = ininsure.insurancestatus;
							insurance.returnfeemode = ininsure.returnfeemode;
//							log.info("returnfeemode-------------------------"+insurance.returnfeemode+"---------"+insurance.netticketid+"-----------"+insurance.insurancestatus);
							if ("1".equals(ininsure.insurancestatus)) {
								String s = "SELECT COUNT(1) FROM `insurancereturn` ins WHERE ins.netticketid=:netticketid ;";
								Query query = JPA.em().createNativeQuery(s);
								query.setParameter("netticketid", ininsure.netticketid);
								if ("0".equals(query.getSingleResult().toString())) {
									InsuranceReturn ir = new InsuranceReturn();
									Date date = new Date();
									ir.netticketid = ininsure.netticketid;
									ir.id = insurance.id;
									ir.insuranceid = insurance.id;
									ir.departorgcode = ininsure.departorgcode;
									ir.departdate = ininsure.departdate;
									ir.returnorgcode = ininsure.returnorgcode;
									ir.premium = insurance.premium;
									ir.returnname = "";
									ir.returntime = ininsure.returntime;
									if (ininsure.returnway!=null) {
										ir.returnway = ininsure.returnway;
									}
									ir.returnip = "";
									ir.ticketoutlets = "";
									ir.createtime = date;
									ir.createby = 0;
									ir.updatetime = date;
									ir.updateby = 0;
									ir.save();
								}
							}
							insurance.save();
						}
					}
				}
				
				JPA.em().getTransaction().commit();
				JPA.em().clear();
			} catch (Exception e) {
				if (!JPA.em().getTransaction().isActive()) {//开启事务
					JPA.em().getTransaction().begin();
				}
				JPA.em().getTransaction().rollback();
				log.error("持久化保险信息数据失败2", e);
				flag = false;
			}
//			String msghashids = msghashidbuffer.toString();
//			if(msghashids.trim().length()>0){
//				finishMsg(msghashids.substring(0, msghashids.length()-1) ,new Date());
//			}
		}catch(Exception e){
			log.error("持久化保险信息数据失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.insurance), messages);
			}
		}
		log.info("insurance消费结束-------------------------"+sdaf.format(new Date()));
	}

	@Override
	public void schedulecancelsyn(List<String> messages) {
		if(!checkCanConsume(Topic.ScheduleCancel.name())){
			return;
		}
		log.info("雪天停运可全额退票停班信息开始消费---------"+messages.size()+"  条---------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka//		
		try{
			Map<Long,ScheduleCancel> schedulecancelmap = new HashMap<Long,ScheduleCancel>();
			StringBuilder hashidsb = new StringBuilder();
			for(String message : messages){
				JsonNode jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				ScheduleCancelinfo sc = (ScheduleCancelinfo) JsonUtil.readJson2Entity(content, ScheduleCancelinfo.class);
				if(sc!=null && (sc.getHashid()==null || sc.getHashid()==0)){
					continue;
				}
				ScheduleCancel schedulecancel = schedulecancelmap.get(sc.getHashid());
				if(schedulecancel == null){
					schedulecancel = new ScheduleCancel();
					schedulecancel.hashid=sc.getHashid();
					schedulecancel.departorgcode = sc.getDepartorgcode();
					schedulecancel.schedulecode = sc.getSchedulecode();
					schedulecancel.departdate = DateUtils.StrToDate(sc.getDepartdate());
					schedulecancel.createtime = DateUtils.strToDate(sc.getCreatetime(),"yyyy-MM-dd HH:mm:ss");
					schedulecancel.remarks = sc.getRemarks();
					schedulecancel.reason = sc.getReason();
					schedulecancelmap.put(sc.getHashid(), schedulecancel);
				}
				hashidsb.append("'").append(sc.getHashid()).append("',");
			}
			if(!"".equals(hashidsb.toString())){
				try {
					String hashids = hashidsb.toString().substring(0, hashidsb.toString().length()-1);
					if (!JPA.em().getTransaction().isActive()) {
						JPA.em().getTransaction().begin();
					}
					String DEL_STAFF_SQL = "delete from schedulecancel  " +
							" where hashid in ("+hashids+")";
					JPA.em().createNativeQuery(DEL_STAFF_SQL).executeUpdate();
					for (ScheduleCancel schedulecancel : schedulecancelmap.values()) {
						JPA.em().persist(schedulecancel);
					}
					JPA.em().getTransaction().commit();
					
				} catch (Exception e) {
					if (!JPA.em().getTransaction().isActive()) {
						JPA.em().getTransaction().begin();
					}
					JPA.em().getTransaction().rollback();
					log.error("消费雪天停运可全额退票停班信息失败2", e);
					flag = false;
				}
			}
		}catch(Exception e){
			log.error("消费雪天停运可全额退票停班信息失败", e);
			flag = false;
		}finally{
			if(!flag){
				ProducerFactory.send(TopicUtil.getTopic(Topic.ScheduleCancel), messages);
			}
			JPA.em().clear();
		}
		log.info("雪天停运可全额退票停班信息消费结束-------------------------------"+sdaf.format(new Date()));
		
	}

	public void stationwebticketmessagesyn(List<String> messages) {
		if(!checkCanConsume("StationWebTicket_xinjiang")){
			return;
		}
		log.info("车站窗口收单信息开始消费---------"+messages.size()+"  条---------------"+sdaf.format(new Date()));
		boolean flag = true;//flag 为false时重发消息到kafka//	
		try{
			Map<String,StationWebTicket> stationwebtickets = new HashMap<String,StationWebTicket>();
			List<String> ordernos = new ArrayList<String>();
			JsonNode jnode;
			StationWebTicket msgticket = null;
			StationWebTicket existticket = null;
			//1.对消息进行去重复并保留最新的消息内容
			for(String message : messages){
				jnode = JsonUtil.getJsonNode(message);
				String content = jnode.get("content").toString();
				msgticket = (StationWebTicket) JsonUtil.readJson2Entity(content, StationWebTicket.class);
				if(stationwebtickets.get(msgticket.netticketid) == null){
					stationwebtickets.put(msgticket.netticketid, msgticket);
				}else{
					existticket = stationwebtickets.get(msgticket.netticketid);
					//如果存在且更新时间比已存在的数据更新时间靠后则使用新消息中的内容
					if(existticket.ticketsellupdatetime.before(msgticket.ticketsellupdatetime) 
							|| (existticket.ticketreturnupdatetime!=null && existticket.ticketreturnupdatetime.before(msgticket.ticketreturnupdatetime))){
						stationwebtickets.put(msgticket.netticketid, msgticket);
					}
				}
				if(!ordernos.contains(msgticket.orderno)){
					ordernos.add(msgticket.orderno);
				}
			}
			//2.将消息内容转换为相应实体对象。
			List<Order> orderresult = JPA.em().createQuery("select o from Order o where o.orderno in :ordernos", Order.class)
					.setParameter("ordernos", ordernos).getResultList();
			List<Ticketsell> ticketsellresult = JPA.em().createQuery("select s from Ticketsell s where s.netticketid in :netticketids", Ticketsell.class)
					.setParameter("netticketids", stationwebtickets.keySet()).getResultList();
			List<Ticketreturn> ticketreturnresult = JPA.em().createQuery("select r from Ticketreturn r where r.netticketid in :netticketids", Ticketreturn.class)
					.setParameter("netticketids", stationwebtickets.keySet()).getResultList();
			List<PayRecord> payrecordlist = JPA.em().createQuery("select p from PayRecord p where p.orderno in :ordernos", PayRecord.class)
					.setParameter("ordernos", ordernos).getResultList();
			
			Set<String> existsorders=new HashSet<String>();
			Set<String> existsticketsells = new HashSet<String>();
			Set<String> existsticketreturns=new HashSet<String>();
			Set<String> existspay=new HashSet<String>();
			
			for(Order o : orderresult){
				existsorders.add(o.orderno);
			}
			for(Ticketsell t : ticketsellresult){
				existsticketsells.add(t.netticketid);
			}
			for(Ticketreturn t : ticketreturnresult){
				existsticketreturns.add(t.getNetticketid());
			}
			for(PayRecord t : payrecordlist){
				existspay.add(t.getOrderno());
			}
			if(!JPA.em().getTransaction().isActive()){
				JPA.em().getTransaction().begin();
			}
			for(String netticketid : stationwebtickets.keySet()){
				msgticket = stationwebtickets.get(netticketid);
				if(!existsorders.contains(msgticket.orderno)){
					saveOrder(msgticket);
					savePayrecord(msgticket);
					existsorders.add(msgticket.orderno);
				}
				if(!existsticketsells.contains(msgticket.netticketid)){
					saveTicketsell(msgticket);
				}else{//如果数据库中已经有车票记录，且车票状态有变化，则更新数据库车票信息。
					for(Ticketsell t : ticketsellresult){
						if(t.netticketid.equals(msgticket.netticketid) && "0".equals(t.ticketstatus) && "1".equals(msgticket.ticketstatus)){
							compareAndUpdateTicketSell(msgticket,t);
						}
					}
				}
			}
			JPA.em().getTransaction().commit();
		}catch(Exception e){
			log.error("消费车站窗口收单信息失败", e);
			flag = false;
		}finally{
			if(!flag){
				log.info("车站窗口收单信息消费失败，重新发送消息");
				ProducerFactory.resend(TopicUtil.getTopic(Topic.StationWebTicket), messages);
			}
			JPA.em().clear();
		}
		log.info("车站窗口收单信息消费结束-------------------------------"+sdaf.format(new Date()));
	}


	private void compareAndUpdateTicketSell(StationWebTicket msgticket,
			Ticketsell t) {
		if("1".equals(msgticket.ticketstatus)){
			t.ticketstatus="1";
			t.departdate = msgticket.departdate;
			t.updatetime = new Date();
			JPA.em().merge(t);
			Ticketreturn ticketreturn= new Ticketreturn();
			ticketreturn.setId(t.id);
			ticketreturn.setNetticketid(msgticket.netticketid);
			ticketreturn.setDepartorgcode(msgticket.departorgcode);
			ticketreturn.setDepartdate(msgticket.departdate);
			ticketreturn.setReturnorgcode(msgticket.sellorgcode);//因数据为站务窗口收单，理解为只有站务自己退票
			ticketreturn.setTicketno(msgticket.ticketno);
			ticketreturn.setTicketprice(new BigDecimal(msgticket.price));
			ticketreturn.setCharges(msgticket.returnfee);
			ticketreturn.setReturnamount(msgticket.price-msgticket.returnfee);
			ticketreturn.setReturnvoucherno("");
			ticketreturn.setReturnname("");
			String returntime = msgticket.returntime;
			if(returntime!=null && !"".equals(returntime)){
				ticketreturn.setReturntime(DateUtils.strToDate(returntime, "yyyy-MM-dd HH:mm:ss"));
			}else{
				ticketreturn.setReturntime(new Date(0));
			}
			ticketreturn.setReturnip("127.0.0.1");
			ticketreturn.setTicketoutlets(msgticket.ticketoutlets);
			ticketreturn.setReturnway("2"); //都是站务退票
			ticketreturn.setReturnfeemode(msgticket.returnfeemode);
			ticketreturn.setTurnoverdetailid(0);
			ticketreturn.setCreatetime(new Date());
			ticketreturn.setCreateby(0);
			ticketreturn.setUpdatetime(ticketreturn.getReturntime());
			ticketreturn.setUpdateby(0);
			//2018-01-02退票方式0,1,2,3,4 退票方式 普通退票 责任退票 特权退票 改签自动退票 单独退保险
			ticketreturn.setReturntype("0");								
			JPA.em().persist(ticketreturn);
		}
	}

	private void savePayrecord(StationWebTicket msgticket) {
		StringBuffer bf = new StringBuffer().append(msgticket.departorgcode).append("-").append(msgticket.orderno).append("-").append(msgticket.paytype).append("-").append(msgticket.paymethod).append("-").append(msgticket.ordertime);
		long hashid = FNVHash.hash64(bf.toString());
		PayRecord payrecord = new PayRecord(msgticket.orderno, msgticket.paytype, msgticket.paymethod,
				msgticket.ordertime, hashid, msgticket.payno, msgticket.departorgcode, "1", msgticket.departdate,
				msgticket.departtime, msgticket.schedulecode, msgticket.seattype, msgticket.reachstationname,
				1, msgticket.amount, msgticket.ordertime, msgticket.payresult, msgticket.payresultinfo,
				"", "", "", msgticket.commercialno, msgticket.unionterminalno, msgticket.bankcardno, msgticket.batchno, null,
				null, null, null, msgticket.serialno, null, msgticket.referno, msgticket.ordertime, new Date());
		JPA.em().persist(payrecord);
	}

	private void saveTicketsell(StationWebTicket msgticket) throws Exception {
		Ticketsell ts = new Ticketsell();

		ts.id = CommonUtils.GetPKID("ticketsell");
		ts.buspark = msgticket.buspark;
		ts.certificateno = msgticket.certificateno;
		ts.certificatetype = msgticket.certificatetype;
		ts.createby = 0l;
		ts.createtime = new Date();
		ts.customername = msgticket.customername;
		ts.departdate = msgticket.departdate;
		ts.departorgcode = msgticket.departorgcode;
		ts.departtime = msgticket.departtime;
		ts.discountrate = 1D;
		ts.distance = 0;
		ts.eticketno = msgticket.eticketno;
		ts.fullprice = msgticket.fullprice;
		ts.islineschedule = "1".equals(msgticket.islineschedule)?true:false;
		ts.takestatus = msgticket.takestatus;
		ts.sellway = msgticket.sellway;
		ts.istakekid = "1".equals(msgticket.istakekid) ? true : false;

		ts.orderno = msgticket.orderno;
		ts.paymethod = msgticket.paymethod;
		ts.phone = msgticket.phone;
		ts.price = msgticket.price;
		ts.reachstationcode = msgticket.reachstationcode;
		ts.schedulecode = msgticket.schedulecode;
		ts.netticketid = msgticket.netticketid;
		ts.seatno = msgticket.seatno;
		ts.seattype = msgticket.seattype;
		ts.sellby = 0l;
		ts.sellip = "127.0.0.1";
		ts.sellorgcode = msgticket.sellorgcode;
		ts.selltime = msgticket.selltime;
		ts.servicefee = msgticket.servicefee==null?0d:msgticket.servicefee;
		ts.ticketentrance = msgticket.ticketentrance;
		ts.ticketno = msgticket.ticketno;
		ts.ticketoutlets = msgticket.ticketoutlets;
		ts.ticketstatus = msgticket.ticketstatus;
		ts.tickettype = msgticket.tickettype;
		ts.toplimitprice = msgticket.toplimitprice;
		ts.updateby = 0l;
		ts.updatetime = msgticket.ticketsellupdatetime;
		ts.verificationcode = msgticket.verificationcode;
		ts.waitingroom = msgticket.waitingroom;
		ts.canreturnmoneyback = "0";
		ts.accountorgcode = msgticket.accountorgcode;
		ts.equipmentproviders = "盛威时代站务";
		ts.seller = "0";
		ts.paymerchantname = msgticket.paymerchantname;
		ts.paymerchantchannelname = msgticket.paymerchantchannelname;
		ts.paymerchantsign = msgticket.paymerchantsign;
		ts.customerid = 0L;
		JPA.em().persist(ts);
		if("1".equals(ts.ticketstatus)){
			Ticketreturn ticketreturn= new Ticketreturn();
			ticketreturn.setId(ts.id);
			ticketreturn.setNetticketid(msgticket.netticketid);
			ticketreturn.setDepartorgcode(msgticket.departorgcode);
			ticketreturn.setDepartdate(msgticket.departdate);
			ticketreturn.setReturnorgcode(msgticket.sellorgcode);//因数据为站务窗口收单，理解为只有站务自己退票
			ticketreturn.setTicketno(msgticket.ticketno);
			ticketreturn.setTicketprice(new BigDecimal(msgticket.price));
			ticketreturn.setCharges(msgticket.returnfee);
			ticketreturn.setReturnamount(msgticket.price-msgticket.returnfee);
			ticketreturn.setReturnvoucherno("");
			ticketreturn.setReturnname("");
			String returntime = msgticket.returntime;
			if(returntime!=null && !"".equals(returntime)){
				ticketreturn.setReturntime(DateUtils.strToDate(returntime, "yyyy-MM-dd HH:mm:ss"));
			}else{
				ticketreturn.setReturntime(new Date(0));
			}
			ticketreturn.setReturnip("127.0.0.1");
			ticketreturn.setTicketoutlets(msgticket.ticketoutlets);
			ticketreturn.setReturnway("2"); //都是站务退票
			ticketreturn.setReturnfeemode(msgticket.returnfeemode);
			ticketreturn.setTurnoverdetailid(0);
			ticketreturn.setCreatetime(new Date(0));
			ticketreturn.setCreateby(0);
			ticketreturn.setUpdatetime(ticketreturn.getReturntime());
			ticketreturn.setUpdateby(0);
			//2018-01-02退票方式0,1,2,3,4 退票方式 普通退票 责任退票 特权退票 改签自动退票 单独退保险
			ticketreturn.setReturntype("0");								
			JPA.em().persist(ticketreturn);
		}
	}

	private void saveOrder(StationWebTicket msgticket) throws Exception {
		Order order = new Order();
		order.id = CommonUtils.GetPKID("orders");
		order.orderno = msgticket.orderno;
		order.amount = msgticket.amount;
		order.createby = 0;
		order.createtime = msgticket.ordertime;
		order.departdate = msgticket.departdate;
		order.departorgcode = msgticket.departorgcode;
		order.departorgname = msgticket.departorgname;
		order.departtime = msgticket.departtime;
		order.lockdelay = 0;
		order.locktime = msgticket.ordertime;
		order.reachstationcode = msgticket.reachstationcode;
		order.reachstationname = msgticket.reachstationname;
		order.schedulecode = msgticket.schedulecode;
		order.seatnum = 1;
		order.seattype = msgticket.seattype;
		order.sellorgcode = msgticket.sellorgcode;
		order.sellorgname = msgticket.sellorgname;
		order.sellway = msgticket.sellway;// 本站售票
		order.updateby = 0;
		order.status = "1";
		order.updatetime = new Date();
		order.save();
	}
}
